chore: Disallow reserve() in clippy to prevent panics (apache#22386)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes apache#123` indicates that this PR will close issue apache#123.
-->

- Closes #.

## Rationale for this change

<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
In apache#22323, we have fixed panic
bugs by replacing `Vec::reserve()` with `Vec::try_reserve()`

This PR enforces a clippy rule to disallow `Vec::reserve()`
project-wise, to prevent future violations. Also it is easy to ignore
this lint with macro, if we can ensure the safety.

## What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->

## Are these changes tested?

<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
They are not testable right now. To trigger the panic, we would need to
construct an Arrow array with more than `i32::MAX` elements, but Arrow
arrays are currently backed by flat vectors.

However, if Arrow supports an encoding such as RLE in the future, it may
become possible to represent such large arrays with constant memory. At
that point, these code paths would become vulnerable, so I think it is
better to fix them now.
## Are there any user-facing changes?

<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->

<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->

Author: 2010YOUY01

clippy.toml

@@ -1,6 +1,7 @@
 disallowed-methods = [
     { path = "tokio::task::spawn", reason = "To provide cancel-safety, use `SpawnedTask::spawn` instead (https://github.com/apache/datafusion/issues/6513)" },
     { path = "tokio::task::spawn_blocking", reason = "To provide cancel-safety, use `SpawnedTask::spawn_blocking` instead (https://github.com/apache/datafusion/issues/6513)" },
+    { path = "std::vec::Vec::reserve", reason = "Use `Vec::try_reserve` so allocation failures can be reported instead of panicking", replacement = "try_reserve" },
 ]
 
 disallowed-types = [

datafusion/functions-aggregate/src/median.rs

@@ -41,7 +41,7 @@ use arrow::datatypes::{
 
 use datafusion_common::types::{NativeType, logical_float64};
 use datafusion_common::{
-    DataFusionError, Result, ScalarValue, assert_eq_or_internal_err,
+    DataFusionError, Result, ScalarValue, assert_eq_or_internal_err, exec_datafusion_err,
     internal_datafusion_err,
 };
 use datafusion_expr::function::StateFieldsArgs;
@@ -282,7 +282,12 @@ impl<T: ArrowNumericType> Accumulator for MedianAccumulator<T> {
 
     fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
         let values = values[0].as_primitive::<T>();
-        self.all_values.reserve(values.len() - values.null_count());
+        let additional = values.len() - values.null_count();
+        self.all_values.try_reserve(additional).map_err(|e| {
+            exec_datafusion_err!(
+                "failed to reserve {additional} values for median accumulator: {e}"
+            )
+        })?;
         self.all_values.extend(values.iter().flatten());
         Ok(())
     }

datafusion/functions-aggregate/src/percentile_cont.rs

@@ -38,7 +38,8 @@ use datafusion_functions_aggregate_common::noop_accumulator::NoopAccumulator;
 
 use crate::min_max::{max_udaf, min_udaf};
 use datafusion_common::{
-    Result, ScalarValue, internal_datafusion_err, utils::take_function_args,
+    Result, ScalarValue, exec_datafusion_err, internal_datafusion_err,
+    utils::take_function_args,
 };
 use datafusion_expr::utils::format_state_name;
 use datafusion_expr::{
@@ -420,7 +421,12 @@ where
 
     fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
         let values = values[0].as_primitive::<T>();
-        self.all_values.reserve(values.len() - values.null_count());
+        let additional = values.len() - values.null_count();
+        self.all_values.try_reserve(additional).map_err(|e| {
+            exec_datafusion_err!(
+                "failed to reserve {additional} values for percentile_cont accumulator: {e}"
+            )
+        })?;
         self.all_values.extend(values.iter().flatten());
         Ok(())
     }

datafusion/functions/src/string/common.rs

@@ -520,6 +520,10 @@ fn case_conversion_utf8view_ascii_inner<F: Fn(&u8) -> u8>(
                     block_size = block_size.saturating_mul(2);
                 }
                 let to_reserve = len.max(block_size as usize);
+                #[expect(
+                    clippy::disallowed_methods,
+                    reason = "StringView's block size bounds growth, so reserve cannot overflow capacity arithmetically. This hot loop intentionally avoids the extra `try_reserve` checks. It remains subject to allocator failure/OOM, which must be managed externally."
+                )]
                 in_progress.reserve(to_reserve);
             }
 

datafusion/functions/src/strings.rs

@@ -703,6 +703,10 @@ impl StringViewArrayBuilder {
         if self.in_progress.capacity() < required_cap {
             self.flush_in_progress();
             let to_reserve = (length as usize).max(self.next_block_size() as usize);
+            #[expect(
+                clippy::disallowed_methods,
+                reason = "StringView's block size bounds growth, so reserve cannot overflow capacity arithmetically. This hot loop intentionally avoids the extra `try_reserve` checks. It remains subject to allocator failure/OOM, which must be managed externally."
+            )]
             self.in_progress.reserve(to_reserve);
         }
 
@@ -730,6 +734,10 @@ impl StringViewArrayBuilder {
         if self.in_progress.capacity() < required_cap {
             self.flush_in_progress();
             let to_reserve = (length as usize).max(self.next_block_size() as usize);
+            #[expect(
+                clippy::disallowed_methods,
+                reason = "StringView's block size bounds growth, so reserve cannot overflow capacity arithmetically. This hot loop intentionally avoids the extra `try_reserve` checks. It remains subject to allocator failure/OOM, which must be managed externally."
+            )]
             self.in_progress.reserve(to_reserve);
         }
     }

datafusion/physical-plan/src/recursive_query.rs

@@ -38,7 +38,9 @@ use arrow::compute::filter_record_batch;
 use arrow::datatypes::{Field, Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
-use datafusion_common::{Result, internal_datafusion_err, not_impl_err};
+use datafusion_common::{
+    Result, exec_datafusion_err, internal_datafusion_err, not_impl_err,
+};
 use datafusion_execution::TaskContext;
 use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
 use datafusion_physical_expr::{EquivalenceProperties, Partitioning};
@@ -480,7 +482,14 @@ impl DistinctDeduplicator {
     /// We also detect duplicates by enforcing that group ids are increasing.
     fn deduplicate(&mut self, batch: &RecordBatch) -> Result<RecordBatch> {
         let size_before = self.group_values.len();
-        self.intern_output_buffer.reserve(batch.num_rows());
+        let additional = batch.num_rows();
+        self.intern_output_buffer
+            .try_reserve(additional)
+            .map_err(|e| {
+                exec_datafusion_err!(
+                    "failed to reserve {additional} recursive query group ids: {e}"
+                )
+            })?;
         self.group_values
             .intern(batch.columns(), &mut self.intern_output_buffer)?;
         let mask = new_groups_mask(&self.intern_output_buffer, size_before);

datafusion/spark/src/function/math/hex.rs

@@ -31,7 +31,7 @@ use datafusion_common::utils::take_function_args;
 use datafusion_common::{
     DataFusionError,
     cast::{as_binary_array, as_fixed_size_binary_array, as_int64_array},
-    exec_err,
+    exec_datafusion_err, exec_err,
 };
 use datafusion_expr::{
     Coercion, ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature,
@@ -178,7 +178,15 @@ where
         if let Some(b) = v {
             let bytes = b.as_ref();
             buffer.clear();
-            buffer.reserve(bytes.len() * 2);
+            let additional = bytes
+                .len()
+                .checked_mul(2)
+                .ok_or_else(|| exec_datafusion_err!("hex output size overflow"))?;
+            buffer.try_reserve(additional).map_err(|e| {
+                exec_datafusion_err!(
+                    "failed to reserve {additional} bytes for hex output: {e}"
+                )
+            })?;
             for &byte in bytes {
                 buffer.extend_from_slice(&lookup[byte as usize]);
             }

datafusion/spark/src/function/math/unhex.rs

@@ -22,7 +22,9 @@ use datafusion_common::cast::{
 };
 use datafusion_common::types::logical_string;
 use datafusion_common::utils::take_function_args;
-use datafusion_common::{DataFusionError, Result, ScalarValue, exec_err};
+use datafusion_common::{
+    DataFusionError, Result, ScalarValue, exec_datafusion_err, exec_err,
+};
 use datafusion_expr::{
     Coercion, ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature,
     TypeSignatureClass, Volatility,
@@ -125,7 +127,12 @@ where
     for v in iter {
         if let Some(s) = v {
             buffer.clear();
-            buffer.reserve(s.as_ref().len().div_ceil(2));
+            let additional = s.as_ref().len().div_ceil(2);
+            buffer.try_reserve(additional).map_err(|e| {
+                exec_datafusion_err!(
+                    "failed to reserve {additional} bytes for unhex output: {e}"
+                )
+            })?;
             if unhex_common(s.as_ref().as_bytes(), &mut buffer) {
                 builder.append_value(&buffer);
             } else {