Revert "toolchain upgrade and error fixes (apache#15625)"

This reverts commit 9d2f049.

Author: gabotechs

datafusion-examples/examples/parquet_index.rs

@@ -685,7 +685,7 @@ fn make_demo_file(path: impl AsRef<Path>, value_range: Range<i32>) -> Result<()>
 
     let num_values = value_range.len();
     let file_names =
-        StringArray::from_iter_values(std::iter::repeat_n(&filename, num_values));
+        StringArray::from_iter_values(std::iter::repeat(&filename).take(num_values));
     let values = Int32Array::from_iter_values(value_range);
     let batch = RecordBatch::try_from_iter(vec![
         ("file_name", Arc::new(file_names) as ArrayRef),

datafusion/common/src/scalar/mod.rs

@@ -27,7 +27,7 @@ use std::convert::Infallible;
 use std::fmt;
 use std::hash::Hash;
 use std::hash::Hasher;
-use std::iter::repeat_n;
+use std::iter::repeat;
 use std::mem::{size_of, size_of_val};
 use std::str::FromStr;
 use std::sync::Arc;
@@ -802,14 +802,12 @@ fn dict_from_scalar<K: ArrowDictionaryKeyType>(
     let values_array = value.to_array_of_size(1)?;
 
     // Create a key array with `size` elements, each of 0
-    let key_array: PrimitiveArray<K> = repeat_n(
-        if value.is_null() {
-            None
-        } else {
-            Some(K::default_value())
-        },
-        size,
-    )
+    let key_array: PrimitiveArray<K> = repeat(if value.is_null() {
+        None
+    } else {
+        Some(K::default_value())
+    })
+    .take(size)
     .collect();
 
     // create a new DictionaryArray
@@ -2191,7 +2189,8 @@ impl ScalarValue {
         scale: i8,
         size: usize,
     ) -> Result<Decimal256Array> {
-        Ok(repeat_n(value, size)
+        Ok(repeat(value)
+            .take(size)
             .collect::<Decimal256Array>()
             .with_precision_and_scale(precision, scale)?)
     }
@@ -2417,59 +2416,69 @@ impl ScalarValue {
             }
             ScalarValue::Utf8(e) => match e {
                 Some(value) => {
-                    Arc::new(StringArray::from_iter_values(repeat_n(value, size)))
+                    Arc::new(StringArray::from_iter_values(repeat(value).take(size)))
                 }
                 None => new_null_array(&DataType::Utf8, size),
             },
             ScalarValue::Utf8View(e) => match e {
                 Some(value) => {
-                    Arc::new(StringViewArray::from_iter_values(repeat_n(value, size)))
+                    Arc::new(StringViewArray::from_iter_values(repeat(value).take(size)))
                 }
                 None => new_null_array(&DataType::Utf8View, size),
             },
             ScalarValue::LargeUtf8(e) => match e {
                 Some(value) => {
-                    Arc::new(LargeStringArray::from_iter_values(repeat_n(value, size)))
+                    Arc::new(LargeStringArray::from_iter_values(repeat(value).take(size)))
                 }
                 None => new_null_array(&DataType::LargeUtf8, size),
             },
             ScalarValue::Binary(e) => match e {
                 Some(value) => Arc::new(
-                    repeat_n(Some(value.as_slice()), size).collect::<BinaryArray>(),
+                    repeat(Some(value.as_slice()))
+                        .take(size)
+                        .collect::<BinaryArray>(),
                 ),
-                None => Arc::new(repeat_n(None::<&str>, size).collect::<BinaryArray>()),
+                None => {
+                    Arc::new(repeat(None::<&str>).take(size).collect::<BinaryArray>())
+                }
             },
             ScalarValue::BinaryView(e) => match e {
                 Some(value) => Arc::new(
-                    repeat_n(Some(value.as_slice()), size).collect::<BinaryViewArray>(),
+                    repeat(Some(value.as_slice()))
+                        .take(size)
+                        .collect::<BinaryViewArray>(),
                 ),
                 None => {
-                    Arc::new(repeat_n(None::<&str>, size).collect::<BinaryViewArray>())
+                    Arc::new(repeat(None::<&str>).take(size).collect::<BinaryViewArray>())
                 }
             },
             ScalarValue::FixedSizeBinary(s, e) => match e {
                 Some(value) => Arc::new(
                     FixedSizeBinaryArray::try_from_sparse_iter_with_size(
-                        repeat_n(Some(value.as_slice()), size),
+                        repeat(Some(value.as_slice())).take(size),
                         *s,
                     )
                     .unwrap(),
                 ),
                 None => Arc::new(
                     FixedSizeBinaryArray::try_from_sparse_iter_with_size(
-                        repeat_n(None::<&[u8]>, size),
+                        repeat(None::<&[u8]>).take(size),
                         *s,
                     )
                     .unwrap(),
                 ),
             },
             ScalarValue::LargeBinary(e) => match e {
                 Some(value) => Arc::new(
-                    repeat_n(Some(value.as_slice()), size).collect::<LargeBinaryArray>(),
+                    repeat(Some(value.as_slice()))
+                        .take(size)
+                        .collect::<LargeBinaryArray>(),
+                ),
+                None => Arc::new(
+                    repeat(None::<&str>)
+                        .take(size)
+                        .collect::<LargeBinaryArray>(),
                 ),
-                None => {
-                    Arc::new(repeat_n(None::<&str>, size).collect::<LargeBinaryArray>())
-                }
             },
             ScalarValue::List(arr) => {
                 Self::list_to_array_of_size(arr.as_ref() as &dyn Array, size)?
@@ -2597,7 +2606,7 @@ impl ScalarValue {
                         child_arrays.push(ar);
                         new_fields.push(field.clone());
                     }
-                    let type_ids = repeat_n(*v_id, size);
+                    let type_ids = repeat(*v_id).take(size);
                     let type_ids = ScalarBuffer::<i8>::from_iter(type_ids);
                     let value_offsets = match mode {
                         UnionMode::Sparse => None,
@@ -2665,7 +2674,7 @@ impl ScalarValue {
     }
 
     fn list_to_array_of_size(arr: &dyn Array, size: usize) -> Result<ArrayRef> {
-        let arrays = repeat_n(arr, size).collect::<Vec<_>>();
+        let arrays = repeat(arr).take(size).collect::<Vec<_>>();
         let ret = match !arrays.is_empty() {
             true => arrow::compute::concat(arrays.as_slice())?,
             false => arr.slice(0, 0),

datafusion/common/src/utils/memory.rs

@@ -25,7 +25,7 @@ use std::mem::size_of;
 /// # Parameters
 /// - `num_elements`: The number of elements expected in the hash table.
 /// - `fixed_size`: A fixed overhead size associated with the collection
-///   (e.g., HashSet or HashTable).
+///    (e.g., HashSet or HashTable).
 /// - `T`: The type of elements stored in the hash table.
 ///
 /// # Details

datafusion/core/src/datasource/listing/table.rs

@@ -1186,7 +1186,7 @@ impl ListingTable {
 /// # Arguments
 /// * `files` - A stream of `Result<PartitionedFile>` items to process
 /// * `limit` - An optional row count limit. If provided, the function will stop collecting files
-///   once the accumulated number of rows exceeds this limit
+///             once the accumulated number of rows exceeds this limit
 /// * `collect_stats` - Whether to collect and accumulate statistics from the files
 ///
 /// # Returns

datafusion/core/src/physical_planner.rs

@@ -1023,12 +1023,18 @@ impl DefaultPhysicalPlanner {
                         // Collect left & right field indices, the field indices are sorted in ascending order
                         let left_field_indices = cols
                             .iter()
-                            .filter_map(|c| left_df_schema.index_of_column(c).ok())
+                            .filter_map(|c| match left_df_schema.index_of_column(c) {
+                                Ok(idx) => Some(idx),
+                                _ => None,
+                            })
                             .sorted()
                             .collect::<Vec<_>>();
                         let right_field_indices = cols
                             .iter()
-                            .filter_map(|c| right_df_schema.index_of_column(c).ok())
+                            .filter_map(|c| match right_df_schema.index_of_column(c) {
+                                Ok(idx) => Some(idx),
+                                _ => None,
+                            })
                             .sorted()
                             .collect::<Vec<_>>();
 

datafusion/core/tests/fuzz_cases/aggregation_fuzzer/context_generator.rs

@@ -43,7 +43,7 @@ use crate::fuzz_cases::aggregation_fuzzer::data_generator::Dataset;
 ///   - `skip_partial parameters`
 ///   - hint `sorted` or not
 ///   - `spilling` or not (TODO, I think a special `MemoryPool` may be needed
-///     to support this)
+///      to support this)
 ///
 pub struct SessionContextGenerator {
     /// Current testing dataset

datafusion/core/tests/fuzz_cases/aggregation_fuzzer/data_generator.rs

@@ -33,12 +33,12 @@ use crate::fuzz_cases::record_batch_generator::{ColumnDescr, RecordBatchGenerato
 ///     when you call `generate` function
 ///         
 ///   - `rows_num_range`, the number of rows in the datasets will be randomly generated
-///     within this range
+///      within this range
 ///
 ///   - `sort_keys`, if `sort_keys` are defined, when you call the `generate` function, the generator
-///     will generate one `base dataset` firstly. Then the `base dataset` will be sorted
-///     based on each `sort_key` respectively. And finally `len(sort_keys) + 1` datasets
-///     will be returned
+///      will generate one `base dataset` firstly. Then the `base dataset` will be sorted
+///      based on each `sort_key` respectively. And finally `len(sort_keys) + 1` datasets
+///      will be returned
 ///
 #[derive(Debug, Clone)]
 pub struct DatasetGeneratorConfig {

datafusion/core/tests/fuzz_cases/aggregation_fuzzer/fuzzer.rs

@@ -270,7 +270,7 @@ impl AggregationFuzzer {
 ///   - `sql`, the selected test sql
 ///
 ///   - `dataset_ref`, the input dataset, store it for error reported when found
-///     the inconsistency between the one for `ctx` and `expected results`.
+///      the inconsistency between the one for `ctx` and `expected results`.
 ///
 struct AggregationFuzzTestTask {
     /// Generated session context in current test case

datafusion/core/tests/memory_limit/mod.rs

@@ -863,10 +863,11 @@ impl Scenario {
                 single_row_batches,
             } => {
                 use datafusion::physical_expr::expressions::col;
-                let batches: Vec<Vec<_>> = std::iter::repeat_n(
-                    maybe_split_batches(dict_batches(), *single_row_batches),
-                    *partitions,
-                )
+                let batches: Vec<Vec<_>> = std::iter::repeat(maybe_split_batches(
+                    dict_batches(),
+                    *single_row_batches,
+                ))
+                .take(*partitions)
                 .collect();
 
                 let schema = batches[0][0].schema();

datafusion/core/tests/parquet/mod.rs

@@ -611,7 +611,7 @@ fn make_bytearray_batch(
     large_binary_values: Vec<&[u8]>,
 ) -> RecordBatch {
     let num_rows = string_values.len();
-    let name: StringArray = std::iter::repeat_n(Some(name), num_rows).collect();
+    let name: StringArray = std::iter::repeat(Some(name)).take(num_rows).collect();
     let service_string: StringArray = string_values.iter().map(Some).collect();
     let service_binary: BinaryArray = binary_values.iter().map(Some).collect();
     let service_fixedsize: FixedSizeBinaryArray = fixedsize_values
@@ -659,7 +659,7 @@ fn make_bytearray_batch(
 /// name | service.name
 fn make_names_batch(name: &str, service_name_values: Vec<&str>) -> RecordBatch {
     let num_rows = service_name_values.len();
-    let name: StringArray = std::iter::repeat_n(Some(name), num_rows).collect();
+    let name: StringArray = std::iter::repeat(Some(name)).take(num_rows).collect();
     let service_name: StringArray = service_name_values.iter().map(Some).collect();
 
     let schema = Schema::new(vec![
@@ -698,31 +698,31 @@ fn make_int_batches_with_null(
                 Int8Array::from_iter(
                     v8.into_iter()
                         .map(Some)
-                        .chain(std::iter::repeat_n(None, null_values)),
+                        .chain(std::iter::repeat(None).take(null_values)),
                 )
                 .to_data(),
             ),
             make_array(
                 Int16Array::from_iter(
                     v16.into_iter()
                         .map(Some)
-                        .chain(std::iter::repeat_n(None, null_values)),
+                        .chain(std::iter::repeat(None).take(null_values)),
                 )
                 .to_data(),
             ),
             make_array(
                 Int32Array::from_iter(
                     v32.into_iter()
                         .map(Some)
-                        .chain(std::iter::repeat_n(None, null_values)),
+                        .chain(std::iter::repeat(None).take(null_values)),
                 )
                 .to_data(),
             ),
             make_array(
                 Int64Array::from_iter(
                     v64.into_iter()
                         .map(Some)
-                        .chain(std::iter::repeat_n(None, null_values)),
+                        .chain(std::iter::repeat(None).take(null_values)),
                 )
                 .to_data(),
             ),