Update workspace to use Rust 1.89

datafusion-examples/examples/parquet_index.rs

@@ -313,7 +313,7 @@ impl Display for ParquetMetadataIndex {
             "ParquetMetadataIndex(last_num_pruned: {})",
             self.last_num_pruned()
         )?;
-        let batches = pretty_format_batches(&[self.index.clone()]).unwrap();
+        let batches = pretty_format_batches(std::slice::from_ref(&self.index)).unwrap();
         write!(f, "{batches}",)
     }
 }

datafusion/catalog/src/async.rs

@@ -737,7 +737,7 @@ mod tests {
         ] {
             let async_provider = MockAsyncCatalogProviderList::default();
             let cached_provider = async_provider
-                .resolve(&[table_ref.clone()], &test_config())
+                .resolve(std::slice::from_ref(table_ref), &test_config())
                 .await
                 .unwrap();
 

datafusion/catalog/src/cte_worktable.rs

@@ -71,7 +71,7 @@ impl TableProvider for CteWorkTable {
         self
     }
 
-    fn get_logical_plan(&self) -> Option<Cow<LogicalPlan>> {
+    fn get_logical_plan(&'_ self) -> Option<Cow<'_, LogicalPlan>> {
         None
     }
 

datafusion/catalog/src/default_table_source.rs

@@ -76,7 +76,7 @@ impl TableSource for DefaultTableSource {
         self.table_provider.supports_filters_pushdown(filter)
     }
 
-    fn get_logical_plan(&self) -> Option<Cow<datafusion_expr::LogicalPlan>> {
+    fn get_logical_plan(&'_ self) -> Option<Cow<'_, datafusion_expr::LogicalPlan>> {
         self.table_provider.get_logical_plan()
     }
 

datafusion/catalog/src/table.rs

@@ -49,7 +49,7 @@ use datafusion_physical_plan::ExecutionPlan;
 /// [`CatalogProvider`]: super::CatalogProvider
 #[async_trait]
 pub trait TableProvider: Debug + Sync + Send {
-    /// Returns the table provider as [`Any`](std::any::Any) so that it can be
+    /// Returns the table provider as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
 
@@ -75,7 +75,7 @@ pub trait TableProvider: Debug + Sync + Send {
     }
 
     /// Get the [`LogicalPlan`] of this table, if available.
-    fn get_logical_plan(&self) -> Option<Cow<LogicalPlan>> {
+    fn get_logical_plan(&'_ self) -> Option<Cow<'_, LogicalPlan>> {
         None
     }
 

datafusion/catalog/src/view.rs

@@ -87,7 +87,7 @@ impl TableProvider for ViewTable {
         self
     }
 
-    fn get_logical_plan(&self) -> Option<Cow<LogicalPlan>> {
+    fn get_logical_plan(&'_ self) -> Option<Cow<'_, LogicalPlan>> {
         Some(Cow::Borrowed(&self.logical_plan))
     }
 

datafusion/common/src/error.rs

@@ -523,7 +523,7 @@ impl DataFusionError {
         }
     }
 
-    pub fn message(&self) -> Cow<str> {
+    pub fn message(&self) -> Cow<'_, str> {
         match *self {
             DataFusionError::ArrowError(ref desc, ref backtrace) => {
                 let backtrace = backtrace.clone().unwrap_or_else(|| "".to_owned());

datafusion/common/src/utils/mod.rs

@@ -260,7 +260,7 @@ pub fn evaluate_partition_ranges(
 /// the identifier by replacing it with two double quotes
 ///
 /// e.g. identifier `tab.le"name` becomes `"tab.le""name"`
-pub fn quote_identifier(s: &str) -> Cow<str> {
+pub fn quote_identifier(s: &str) -> Cow<'_, str> {
     if needs_quotes(s) {
         Cow::Owned(format!("\"{}\"", s.replace('"', "\"\"")))
     } else {

datafusion/core/src/dataframe/mod.rs

@@ -2425,7 +2425,7 @@ impl TableProvider for DataFrameTableProvider {
         self
     }
 
-    fn get_logical_plan(&self) -> Option<Cow<LogicalPlan>> {
+    fn get_logical_plan(&self) -> Option<Cow<'_, LogicalPlan>> {
         Some(Cow::Borrowed(&self.plan))
     }
 

datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs

@@ -320,7 +320,7 @@ async fn run_aggregate_test(input1: Vec<RecordBatch>, group_by_columns: Vec<&str
     .unwrap();
 
     let running_source = DataSourceExec::from_data_source(
-        MemorySourceConfig::try_new(&[input1.clone()], schema.clone(), None)
+        MemorySourceConfig::try_new(std::slice::from_ref(&input1), schema.clone(), None)
             .unwrap()
             .try_with_sort_information(vec![sort_keys.into()])
             .unwrap(),

datafusion/core/tests/fuzz_cases/join_fuzz.rs

@@ -477,12 +477,18 @@ impl JoinFuzzTestCase {
     fn left_right(&self) -> (Arc<DataSourceExec>, Arc<DataSourceExec>) {
         let schema1 = self.input1[0].schema();
         let schema2 = self.input2[0].schema();
-        let left =
-            MemorySourceConfig::try_new_exec(&[self.input1.clone()], schema1, None)
-                .unwrap();
-        let right =
-            MemorySourceConfig::try_new_exec(&[self.input2.clone()], schema2, None)
-                .unwrap();
+        let left = MemorySourceConfig::try_new_exec(
+            std::slice::from_ref(&self.input1),
+            schema1,
+            None,
+        )
+        .unwrap();
+        let right = MemorySourceConfig::try_new_exec(
+            std::slice::from_ref(&self.input2),
+            schema2,
+            None,
+        )
+        .unwrap();
         (left, right)
     }
 

datafusion/core/tests/macro_hygiene/mod.rs

@@ -83,6 +83,7 @@ mod config_field {
 
         impl std::error::Error for E {}
 
+        #[allow(dead_code)]
         struct S;
 
         impl std::str::FromStr for S {

datafusion/core/tests/parquet/file_statistics.rs

@@ -72,7 +72,7 @@ async fn check_stats_precision_with_filter_pushdown() {
     // source operator after the appropriate optimizer pass.
     let filter_expr = Expr::gt(col("id"), lit(1));
     let exec_with_filter = table
-        .scan(&state, None, &[filter_expr.clone()], None)
+        .scan(&state, None, std::slice::from_ref(&filter_expr), None)
         .await
         .unwrap();
 

datafusion/core/tests/physical_optimizer/partition_statistics.rs

@@ -746,8 +746,11 @@ mod test {
         for (i, partition_stream) in partitions.into_iter().enumerate() {
             let batches: Vec<RecordBatch> = partition_stream.try_collect().await?;
             let actual = plan.partition_statistics(Some(i))?;
-            let expected =
-                compute_record_batch_statistics(&[batches.clone()], &schema, None);
+            let expected = compute_record_batch_statistics(
+                std::slice::from_ref(&batches),
+                &schema,
+                None,
+            );
             assert_eq!(actual, expected);
             all_batches.push(batches);
         }

datafusion/datasource-avro/src/avro_to_arrow/arrow_array_reader.rs

@@ -1046,7 +1046,7 @@ mod test {
     use std::fs::File;
     use std::sync::Arc;
 
-    fn build_reader(name: &str, batch_size: usize) -> Reader<File> {
+    fn build_reader(name: &'_ str, batch_size: usize) -> Reader<'_, File> {
         let testdata = datafusion_common::test_util::arrow_test_data();
         let filename = format!("{testdata}/avro/{name}");
         let builder = ReaderBuilder::new()

datafusion/datasource-avro/src/avro_to_arrow/reader.rs

@@ -195,7 +195,7 @@ mod tests {
     use arrow::datatypes::{DataType, Field};
     use std::fs::File;
 
-    fn build_reader(name: &str, projection: Option<Vec<String>>) -> Reader<File> {
+    fn build_reader(name: &'_ str, projection: Option<Vec<String>>) -> Reader<'_, File> {
         let testdata = datafusion_common::test_util::arrow_test_data();
         let filename = format!("{testdata}/avro/{name}");
         let mut builder = ReaderBuilder::new().read_schema().with_batch_size(64);

datafusion/datasource-parquet/src/opener.rs

@@ -237,11 +237,11 @@ impl FileOpener for ParquetOpener {
                 )?;
             }
 
-            if coerce_int96.is_some() {
+            if let Some(ref coerce) = coerce_int96 {
                 if let Some(merged) = coerce_int96_to_resolution(
                     reader_metadata.parquet_schema(),
                     &physical_file_schema,
-                    &(coerce_int96.unwrap()),
+                    coerce,
                 ) {
                     physical_file_schema = Arc::new(merged);
                     options = options.with_schema(Arc::clone(&physical_file_schema));

datafusion/datasource/src/file_format.rs

@@ -48,7 +48,7 @@ pub const DEFAULT_SCHEMA_INFER_MAX_RECORD: usize = 1000;
 /// [`TableProvider`]: https://docs.rs/datafusion/latest/datafusion/catalog/trait.TableProvider.html
 #[async_trait]
 pub trait FileFormat: Send + Sync + fmt::Debug {
-    /// Returns the table provider as [`Any`](std::any::Any) so that it can be
+    /// Returns the table provider as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
 

datafusion/datasource/src/file_scan_config.rs

@@ -1982,7 +1982,7 @@ mod tests {
             );
             let result = FileScanConfig::split_groups_by_statistics(
                 &table_schema,
-                &[partitioned_files.clone()],
+                std::slice::from_ref(&partitioned_files),
                 &sort_order,
             );
             let results_by_name = result

datafusion/datasource/src/memory.rs

@@ -650,7 +650,7 @@ impl RePartition {
 
 impl PartialOrd for RePartition {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        Some(self.row_count.cmp(&other.row_count))
+        Some(self.cmp(other))
     }
 }
 

datafusion/datasource/src/sink.rs

@@ -46,7 +46,7 @@ use futures::StreamExt;
 /// output.
 #[async_trait]
 pub trait DataSink: DisplayAs + Debug + Send + Sync {
-    /// Returns the data sink as [`Any`](std::any::Any) so that it can be
+    /// Returns the data sink as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
 

datafusion/datasource/src/statistics.rs

@@ -57,12 +57,12 @@ impl MinMaxStatistics {
 
     /// Min value at index
     #[allow(unused)]
-    pub fn min(&self, idx: usize) -> Row {
+    pub fn min(&'_ self, idx: usize) -> Row<'_> {
         self.min_by_sort_order.row(idx)
     }
 
     /// Max value at index
-    pub fn max(&self, idx: usize) -> Row {
+    pub fn max(&'_ self, idx: usize) -> Row<'_> {
         self.max_by_sort_order.row(idx)
     }
 

datafusion/expr/src/table_source.rs

@@ -121,7 +121,7 @@ pub trait TableSource: Sync + Send {
     /// Get the Logical plan of this table provider, if available.
     ///
     /// For example, a view may have a logical plan, but a CSV file does not.
-    fn get_logical_plan(&self) -> Option<Cow<LogicalPlan>> {
+    fn get_logical_plan(&'_ self) -> Option<Cow<'_, LogicalPlan>> {
         None
     }
 

datafusion/functions-aggregate/benches/array_agg.rs

@@ -45,7 +45,7 @@ fn merge_batch_bench(c: &mut Criterion, name: &str, values: ArrayRef) {
             black_box(
                 ArrayAggAccumulator::try_new(&list_item_data_type, false)
                     .unwrap()
-                    .merge_batch(&[values.clone()])
+                    .merge_batch(std::slice::from_ref(&values))
                     .unwrap(),
             )
         })

datafusion/functions-aggregate/benches/count.rs

@@ -82,7 +82,7 @@ fn convert_to_state_bench(
         b.iter(|| {
             black_box(
                 accumulator
-                    .convert_to_state(&[values.clone()], opt_filter)
+                    .convert_to_state(std::slice::from_ref(&values), opt_filter)
                     .unwrap(),
             )
         })
@@ -125,7 +125,11 @@ fn count_benchmark(c: &mut Criterion) {
     c.bench_function("count low cardinality dict 20% nulls, no filter", |b| {
         b.iter(|| {
             #[allow(clippy::unit_arg)]
-            black_box(accumulator.update_batch(&[values.clone()]).unwrap())
+            black_box(
+                accumulator
+                    .update_batch(std::slice::from_ref(&values))
+                    .unwrap(),
+            )
         })
     });
 }

datafusion/functions-aggregate/benches/sum.rs

@@ -56,7 +56,7 @@ fn convert_to_state_bench(
         b.iter(|| {
             black_box(
                 accumulator
-                    .convert_to_state(&[values.clone()], opt_filter)
+                    .convert_to_state(std::slice::from_ref(&values), opt_filter)
                     .unwrap(),
             )
         })

datafusion/functions/src/datetime/common.rs

@@ -412,8 +412,8 @@ where
                     }?;
 
                     let r = op(x, v);
-                    if r.is_ok() {
-                        val = Some(Ok(op2(r.unwrap())));
+                    if let Ok(inner) = r {
+                        val = Some(Ok(op2(inner)));
                         break;
                     } else {
                         val = Some(r);

datafusion/functions/src/datetime/to_local_time.rs

@@ -372,7 +372,7 @@ impl ScalarUDFImpl for ToLocalTimeFunc {
     ) -> Result<ColumnarValue> {
         let [time_value] = take_function_args(self.name(), args.args)?;
 
-        self.to_local_time(&[time_value.clone()])
+        self.to_local_time(std::slice::from_ref(&time_value))
     }
 
     fn coerce_types(&self, arg_types: &[DataType]) -> Result<Vec<DataType>> {

datafusion/functions/src/macros.rs

@@ -73,6 +73,7 @@ macro_rules! export_functions {
 #[macro_export]
 macro_rules! make_udf_function {
     ($UDF:ty, $NAME:ident) => {
+        #[allow(rustdoc::redundant_explicit_links)]
         #[doc = concat!("Return a [`ScalarUDF`](datafusion_expr::ScalarUDF) implementation of ", stringify!($NAME))]
         pub fn $NAME() -> std::sync::Arc<datafusion_expr::ScalarUDF> {
             // Singleton instance of the function

datafusion/functions/src/math/log.rs

@@ -648,7 +648,7 @@ mod tests {
 
         // Test log(num)
         for order in orders.iter().cloned() {
-            let result = log.output_ordering(&[order.clone()]).unwrap();
+            let result = log.output_ordering(std::slice::from_ref(&order)).unwrap();
             assert_eq!(result, order.sort_properties);
         }
 

datafusion/functions/src/unicode/lpad.rs

@@ -204,11 +204,15 @@ where
     V2: StringArrayType<'a>,
     T: OffsetSizeTrait,
 {
-    let array = if fill_array.is_none() {
+    let array = if let Some(fill_array) = fill_array {
         let mut builder: GenericStringBuilder<T> = GenericStringBuilder::new();
 
-        for (string, length) in string_array.iter().zip(length_array.iter()) {
-            if let (Some(string), Some(length)) = (string, length) {
+        for ((string, length), fill) in string_array
+            .iter()
+            .zip(length_array.iter())
+            .zip(fill_array.iter())
+        {
+            if let (Some(string), Some(length), Some(fill)) = (string, length, fill) {
                 if length > i32::MAX as i64 {
                     return exec_err!("lpad requested length {length} too large");
                 }
@@ -220,10 +224,17 @@ where
                 }
 
                 let graphemes = string.graphemes(true).collect::<Vec<&str>>();
+                let fill_chars = fill.chars().collect::<Vec<char>>();
+
                 if length < graphemes.len() {
                     builder.append_value(graphemes[..length].concat());
+                } else if fill_chars.is_empty() {
+                    builder.append_value(string);
                 } else {
-                    builder.write_str(" ".repeat(length - graphemes.len()).as_str())?;
+                    for l in 0..length - graphemes.len() {
+                        let c = *fill_chars.get(l % fill_chars.len()).unwrap();
+                        builder.write_char(c)?;
+                    }
                     builder.write_str(string)?;
                     builder.append_value("");
                 }
@@ -236,12 +247,8 @@ where
     } else {
         let mut builder: GenericStringBuilder<T> = GenericStringBuilder::new();
 
-        for ((string, length), fill) in string_array
-            .iter()
-            .zip(length_array.iter())
-            .zip(fill_array.unwrap().iter())
-        {
-            if let (Some(string), Some(length), Some(fill)) = (string, length, fill) {
+        for (string, length) in string_array.iter().zip(length_array.iter()) {
+            if let (Some(string), Some(length)) = (string, length) {
                 if length > i32::MAX as i64 {
                     return exec_err!("lpad requested length {length} too large");
                 }
@@ -253,17 +260,10 @@ where
                 }
 
                 let graphemes = string.graphemes(true).collect::<Vec<&str>>();
-                let fill_chars = fill.chars().collect::<Vec<char>>();
-
                 if length < graphemes.len() {
                     builder.append_value(graphemes[..length].concat());
-                } else if fill_chars.is_empty() {
-                    builder.append_value(string);
                 } else {
-                    for l in 0..length - graphemes.len() {
-                        let c = *fill_chars.get(l % fill_chars.len()).unwrap();
-                        builder.write_char(c)?;
-                    }
+                    builder.write_str(" ".repeat(length - graphemes.len()).as_str())?;
                     builder.write_str(string)?;
                     builder.append_value("");
                 }