chore: Update Rust toolchain to 1.98.1

Cargo.toml

@@ -283,7 +283,6 @@ enum_glob_use = "allow"                      # 98 hits
 explicit_into_iter_loop = "allow"            # 55 hits
 explicit_iter_loop = "allow"                 # 189 hits
 float_cmp = "allow"                          # 8 hits; exact float comparisons are often intentional here
-from_iter_instead_of_collect = "allow"       # 51 hits
 if_not_else = "allow"                        # 133 hits
 ignored_unit_patterns = "allow"              # 52 hits
 implicit_clone = "allow"                     # 198 hits

datafusion-cli/src/helper.rs

@@ -161,7 +161,7 @@ impl Completer for CliHelper {
         if is_open_quote_for_location(line, pos) {
             self.completer.complete(line, pos, ctx)
         } else {
-            Ok((0, Vec::with_capacity(0)))
+            Ok((0, Vec::new()))
         }
     }
 }

datafusion-cli/src/object_storage.rs

@@ -58,6 +58,7 @@ use object_store::aws::resolve_bucket_region;
 #[cfg(test)]
 #[expect(
     clippy::unused_async,
+    clippy::result_large_err,
     reason = "matches object_store::aws::resolve_bucket_region"
 )]
 async fn resolve_bucket_region(

datafusion-cli/src/object_storage/instrumented.rs

@@ -143,6 +143,10 @@ pub struct InstrumentedObjectStore {
     requests: Arc<Mutex<Vec<RequestDetails>>>,
 }
 
+#[expect(
+    clippy::result_large_err,
+    reason = "error type is dictated by the object_store API"
+)]
 impl InstrumentedObjectStore {
     /// Returns a new [`InstrumentedObjectStore`] that wraps the provided [`ObjectStore`]
     fn new(object_store: Arc<dyn ObjectStore>, instrument_mode: AtomicU8) -> Self {

datafusion-examples/examples/data_io/object_store_spill.rs

@@ -242,6 +242,10 @@ struct ObjectStoreSpillWriter {
 }
 
 impl ObjectStoreSpillWriter {
+    #[expect(
+        clippy::result_large_err,
+        reason = "error type is dictated by the object_store API"
+    )]
     async fn flush_part(&mut self) -> object_store::Result<()> {
         if self.buffer.is_empty() {
             return Ok(());

datafusion-examples/examples/data_io/remote_catalog.rs

@@ -130,15 +130,15 @@ struct RemoteCatalogInterface {}
 
 impl RemoteCatalogInterface {
     /// Establish a connection to the remote catalog
-    #[expect(clippy::unused_async)]
+    #[expect(clippy::unused_async, clippy::unused_async_trait_impl)]
     pub async fn connect() -> Result<Self> {
         // In a real implementation this method might connect to a remote
         // catalog, validate credentials, cache basic information, etc
         Ok(Self {})
     }
 
     /// Fetches information for a specific table
-    #[expect(clippy::unused_async)]
+    #[expect(clippy::unused_async, clippy::unused_async_trait_impl)]
     pub async fn table_info(&self, name: &str) -> Result<Option<SchemaRef>> {
         if name != "remote_table" {
             return Ok(None);
@@ -157,7 +157,7 @@ impl RemoteCatalogInterface {
     }
 
     /// Fetches data for a table from a remote data source
-    #[expect(clippy::unused_async)]
+    #[expect(clippy::unused_async, clippy::unused_async_trait_impl)]
     pub async fn read_data(&self, name: &str) -> Result<SendableRecordBatchStream> {
         if name != "remote_table" {
             return plan_err!("Remote table not found: {}", name);

datafusion-examples/examples/flight/client.rs

@@ -49,7 +49,7 @@ pub async fn client() -> Result<(), Box<dyn std::error::Error>> {
     let request = tonic::Request::new(FlightDescriptor {
         r#type: flight_descriptor::DescriptorType::Path as i32,
         cmd: Default::default(),
-        path: vec![format!("{}", parquet_temp.path_str()?)],
+        path: vec![parquet_temp.path_str()?.to_string()],
     });
 
     let schema_result = client.get_schema(request).await?.into_inner();

datafusion/common/src/hash_utils.rs

@@ -1285,10 +1285,6 @@ mod tests {
     use std::hash::{BuildHasherDefault, Hasher};
     use std::sync::Arc;
 
-    use arrow::array::*;
-    #[cfg(not(feature = "force_hash_collisions"))]
-    use arrow::datatypes::*;
-
     use super::*;
 
     #[cfg(not(feature = "force_hash_collisions"))]

datafusion/common/src/scalar/mod.rs

@@ -7921,7 +7921,7 @@ mod tests {
 
         #[allow(clippy::allow_attributes, clippy::mutable_key_type)]
         // ScalarValue has interior mutability but is intentionally used as hash key
-        let mut s = HashSet::with_capacity(0);
+        let mut s = HashSet::new();
         // do NOT clone `sv` here because this may shrink the vector capacity
         s.insert(v.pop().unwrap());
         // hashsets may easily grow during insert, so capacity is dynamic

datafusion/common/src/utils/hex.rs

@@ -153,8 +153,8 @@ pub fn encode_bytes_to_slice(bytes: &[u8], case: HexCase, out: &mut [u8]) -> Res
         );
     }
     let lookup = case.lookup();
-    for (&b, chunk) in bytes.iter().zip(out.chunks_exact_mut(2)) {
-        chunk.copy_from_slice(&lookup[b as usize]);
+    for (&b, chunk) in bytes.iter().zip(out.as_chunks_mut::<2>().0) {
+        *chunk = lookup[b as usize];
     }
     Ok(())
 }

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -1157,7 +1157,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
     // Now check what our filter looks like
     #[cfg(not(feature = "force_hash_collisions"))]
     insta::assert_snapshot!(
-        format!("{}", format_plan_for_test(&plan)),
+        format_plan_for_test(&plan),
         @r"
     - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
     -   CoalescePartitionsExec
@@ -1175,7 +1175,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
     // joins or any scenario where all build-side data naturally lands in one partition.
     #[cfg(feature = "force_hash_collisions")]
     insta::assert_snapshot!(
-        format!("{}", format_plan_for_test(&plan)),
+        format_plan_for_test(&plan),
         @r"
     - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
     -   CoalescePartitionsExec
@@ -1373,7 +1373,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_range_partitioned() {
 
     // Now check what our filter looks like
     insta::assert_snapshot!(
-        format!("{}", format_plan_for_test(&plan)),
+        format_plan_for_test(&plan),
         @r"
     - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
     -   CoalescePartitionsExec
@@ -1502,7 +1502,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_collect_left() {
 
     // Now check what our filter looks like
     insta::assert_snapshot!(
-        format!("{}", format_plan_for_test(&plan)),
+        format_plan_for_test(&plan),
         @r"
     - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
     -   CoalescePartitionsExec

datafusion/datasource/src/boundary_stream.rs

@@ -84,6 +84,10 @@ pub struct AlignedBoundaryStream {
 }
 
 /// Fetch a bounded byte range from `store` and return it as a stream
+#[expect(
+    clippy::result_large_err,
+    reason = "error type is dictated by the object_store API"
+)]
 async fn get_stream(
     store: Arc<dyn ObjectStore>,
     location: object_store::path::Path,
@@ -148,6 +152,10 @@ impl AlignedBoundaryStream {
     /// newline is not found within that window, `ScanningLastTerminator`
     /// automatically issues additional `END_SCAN_LOOKAHEAD`-sized GETs
     /// via `store` until the newline is found or EOF is reached.
+    #[expect(
+        clippy::result_large_err,
+        reason = "error type is dictated by the object_store API"
+    )]
     pub async fn new(
         store: Arc<dyn ObjectStore>,
         location: object_store::path::Path,

datafusion/expr/src/logical_plan/display.rs

@@ -426,7 +426,7 @@ impl<'a, 'b> PgJsonVisitor<'a, 'b> {
                 json!({
                     "Node Type": "CopyTo",
                     "Output URL": output_url,
-                    "File Type": format!("{}", file_type.get_ext()),
+                    "File Type": file_type.get_ext(),
                     "Options": op_str
                 })
             }
@@ -490,7 +490,7 @@ impl<'a, 'b> PgJsonVisitor<'a, 'b> {
                     "Node Type": format!("{} Join", join_type),
                     "Join Constraint": format!("{:?}", join_constraint),
                     "Join Keys": join_expr.join(", "),
-                    "Filter": format!("{}", filter_expr)
+                    "Filter": filter_expr.to_string()
                 })
             }
             LogicalPlan::AsOfJoin(AsOfJoin {

datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/accumulate.rs

@@ -443,14 +443,13 @@ pub fn accumulate<T, F>(
             let nulls = values.nulls().unwrap();
             // This is based on (ahem, COPY/PASTE) arrow::compute::aggregate::sum
             // iterate over in chunks of 64 bits for more efficient null checking
-            let group_indices_chunks = group_indices.chunks_exact(64);
-            let data_chunks = data.chunks_exact(64);
+            let (group_indices_chunks, group_indices_remainder) =
+                group_indices.as_chunks::<64>();
+            let (data_chunks, data_remainder) = data.as_chunks::<64>();
             let bit_chunks = nulls.inner().bit_chunks();
 
-            let group_indices_remainder = group_indices_chunks.remainder();
-            let data_remainder = data_chunks.remainder();
-
             group_indices_chunks
+                .iter()
                 .zip(data_chunks)
                 .zip(bit_chunks.iter())
                 .for_each(|((group_index_chunk, data_chunk), mask)| {
@@ -605,13 +604,12 @@ pub fn accumulate_indices<F>(
         }
         (None, Some(filter)) => {
             debug_assert_eq!(filter.len(), group_indices.len());
-            let group_indices_chunks = group_indices.chunks_exact(64);
+            let (group_indices_chunks, group_indices_remainder) =
+                group_indices.as_chunks::<64>();
             let filter_validity = filter_to_validity(filter);
             let bit_chunks = filter_validity.bit_chunks();
 
-            let group_indices_remainder = group_indices_chunks.remainder();
-
-            group_indices_chunks.zip(bit_chunks.iter()).for_each(
+            group_indices_chunks.iter().zip(bit_chunks.iter()).for_each(
                 |(group_index_chunk, mask)| {
                     // index_mask has value 1 << i in the loop
                     let mut index_mask = 1;
@@ -642,12 +640,11 @@ pub fn accumulate_indices<F>(
             debug_assert_eq!(valids.len(), group_indices.len());
             // This is based on (ahem, COPY/PASTA) arrow::compute::aggregate::sum
             // iterate over in chunks of 64 bits for more efficient null checking
-            let group_indices_chunks = group_indices.chunks_exact(64);
+            let (group_indices_chunks, group_indices_remainder) =
+                group_indices.as_chunks::<64>();
             let bit_chunks = valids.inner().bit_chunks();
 
-            let group_indices_remainder = group_indices_chunks.remainder();
-
-            group_indices_chunks.zip(bit_chunks.iter()).for_each(
+            group_indices_chunks.iter().zip(bit_chunks.iter()).for_each(
                 |(group_index_chunk, mask)| {
                     // index_mask has value 1 << i in the loop
                     let mut index_mask = 1;
@@ -679,14 +676,14 @@ pub fn accumulate_indices<F>(
             debug_assert_eq!(filter.len(), group_indices.len());
             debug_assert_eq!(valids.len(), group_indices.len());
 
-            let group_indices_chunks = group_indices.chunks_exact(64);
+            let (group_indices_chunks, group_indices_remainder) =
+                group_indices.as_chunks::<64>();
             let valid_bit_chunks = valids.inner().bit_chunks();
             let filter_validity = filter_to_validity(filter);
             let filter_bit_chunks = filter_validity.bit_chunks();
 
-            let group_indices_remainder = group_indices_chunks.remainder();
-
             group_indices_chunks
+                .iter()
                 .zip(valid_bit_chunks.iter())
                 .zip(filter_bit_chunks.iter())
                 .for_each(|((group_index_chunk, valid_mask), filter_mask)| {

datafusion/functions-aggregate/src/approx_distinct.rs

@@ -359,9 +359,8 @@ impl GroupHll {
                 );
             }
             let mut delta = 0;
-            for chunk in bytes.chunks_exact(size_of::<u64>()) {
-                let h = u64::from_le_bytes(chunk.try_into().unwrap());
-                delta += self.add_hash(h);
+            for chunk in bytes.as_chunks::<{ size_of::<u64>() }>().0 {
+                delta += self.add_hash(u64::from_le_bytes(*chunk));
             }
             Ok(delta)
         }

datafusion/functions/src/core/getfield.rs

@@ -302,7 +302,7 @@ fn simplify_get_field_over_struct_constructor(args: &[Expr]) -> Option<Expr> {
             return None;
         }
         let mut matched = None;
-        for pair in ctor_args.chunks_exact(2) {
+        for [name_expr, value_expr] in ctor_args.as_chunks::<2>().0 {
             // Every name must be a literal string: a non-literal name appearing
             // *before* the first match could evaluate to `field_name` at runtime
             // and become the real first match (Arrow's `column_by_name` returns
@@ -313,13 +313,13 @@ fn simplify_get_field_over_struct_constructor(args: &[Expr]) -> Option<Expr> {
             // — it can never precede the first match — so bailing there is a
             // deliberate approximation we accept to keep this check simple, not a
             // correctness requirement.
-            let Expr::Literal(name, _) = &pair[0] else {
+            let Expr::Literal(name, _) = name_expr else {
                 return None;
             };
             let name = name.try_as_str().flatten()?;
             // `column_by_name` resolves to the first match, so do the same.
             if matched.is_none() && name == field_name {
-                matched = Some(&pair[1]);
+                matched = Some(value_expr);
             }
         }
         matched?.clone()

datafusion/functions/src/core/named_struct.rs

@@ -162,8 +162,10 @@ impl ScalarUDFImpl for NamedStructFunc {
 
         let values: Vec<ColumnarValue> = args
             .args
-            .chunks_exact(2)
-            .map(|chunk| chunk[1].clone())
+            .as_chunks::<2>()
+            .0
+            .iter()
+            .map(|[_name, value]| value.clone())
             .collect();
         let arrays = ColumnarValue::values_to_arrays(&values)?;
         Ok(ColumnarValue::Array(Arc::new(StructArray::new(

datafusion/physical-plan/src/repartition/distributor_channels.rs

@@ -232,7 +232,7 @@ impl<T> Future for SendFuture<'_, T> {
                 this.gate.decr_empty_channels();
                 guard_channel_state.take_recv_wakers()
             } else {
-                Vec::with_capacity(0)
+                Vec::new()
             }
         };
 
@@ -316,10 +316,10 @@ impl<T> Future for RecvFuture<'_, T> {
                         if this.gate.empty_channels.load(Ordering::SeqCst) > 0 {
                             guard.take().unwrap_or_default()
                         } else {
-                            Vec::with_capacity(0)
+                            Vec::new()
                         }
                     } else {
-                        Vec::with_capacity(0)
+                        Vec::new()
                     };
 
                     drop(guard_channel_state);
@@ -439,7 +439,7 @@ impl Gate {
 
                 wake
             } else {
-                Vec::with_capacity(0)
+                Vec::new()
             }
         };
 

datafusion/sql/src/unparser/expr.rs

@@ -406,20 +406,18 @@ impl Unparser<'_> {
                     ..
                 } = &agg.params;
 
-                let args_to_use;
-                let within_group;
-
                 // if this is a WITHIN GROUP aggregate, skip the prepended arg
-                if agg.func.supports_within_group_clause() && !order_by.is_empty() {
-                    args_to_use = self.function_args_to_sql(&args[1..])?;
-                    within_group = order_by
-                        .iter()
-                        .map(|sort_expr| self.sort_to_sql(sort_expr))
-                        .collect::<Result<Vec<ast::OrderByExpr>>>()?;
-                } else {
-                    args_to_use = self.function_args_to_sql(args)?;
-                    within_group = Vec::new();
-                }
+                let (args_to_use, within_group) =
+                    if agg.func.supports_within_group_clause() && !order_by.is_empty() {
+                        let args_to_use = self.function_args_to_sql(&args[1..])?;
+                        let within_group = order_by
+                            .iter()
+                            .map(|sort_expr| self.sort_to_sql(sort_expr))
+                            .collect::<Result<Vec<ast::OrderByExpr>>>()?;
+                        (args_to_use, within_group)
+                    } else {
+                        (self.function_args_to_sql(args)?, Vec::new())
+                    };
 
                 let filter = match filter {
                     Some(filter) => Some(Box::new(self.expr_to_sql_inner(filter)?)),
@@ -742,16 +740,18 @@ impl Unparser<'_> {
         );
 
         let args = args
-            .chunks_exact(2)
-            .map(|chunk| {
-                let key = match &chunk[0] {
+            .as_chunks::<2>()
+            .0
+            .iter()
+            .map(|[name, value]| {
+                let key = match name {
                     Expr::Literal(ScalarValue::Utf8(Some(s)), _) => self.new_ident_quoted_if_needs(s.to_string()),
-                    _ => return internal_err!("named_struct expects even arguments to be strings, but received: {:?}", &chunk[0])
+                    _ => return internal_err!("named_struct expects even arguments to be strings, but received: {name:?}")
                 };
 
                 Ok(ast::DictionaryField {
                     key,
-                    value: Box::new(self.expr_to_sql_with_nesting(&chunk[1])?),
+                    value: Box::new(self.expr_to_sql_with_nesting(value)?),
                 })
             })
             .collect::<Result<Vec<_>>>()?;