Deprecate adjust_output_array in favor of PrimitiveArray::with_data_type

Author: alamb

datafusion/functions-aggregate-common/src/utils.rs

@@ -48,6 +48,7 @@ pub fn get_accum_scalar_values_as_arrays(
 /// Since `Decimal128Arrays` created from `Vec<NativeType>` have
 /// default precision and scale, this function adjusts the output to
 /// match `data_type`, if necessary
+#[deprecated(since = "44.0.0", note = "use PrimitiveArray::with_datatype")]
 pub fn adjust_output_array(data_type: &DataType, array: ArrayRef) -> Result<ArrayRef> {
     let array = match data_type {
         DataType::Decimal128(p, s) => Arc::new(

datafusion/physical-expr/src/aggregate.rs

@@ -28,8 +28,10 @@ pub(crate) mod stats {
     pub use datafusion_functions_aggregate_common::stats::StatsType;
 }
 pub mod utils {
+    #[allow(deprecated)] // allow adjust_output_array
     pub use datafusion_functions_aggregate_common::utils::{
-        adjust_output_array, get_accum_scalar_values_as_arrays, get_sort_options,
+        adjust_output_array,
+        get_accum_scalar_values_as_arrays, get_sort_options,
         ordering_fields, DecimalAverager, Hashable,
     };
 }

datafusion/physical-plan/src/aggregates/topk/heap.rs

@@ -20,11 +20,11 @@
 use arrow::datatypes::i256;
 use arrow_array::cast::AsArray;
 use arrow_array::{downcast_primitive, ArrayRef, ArrowPrimitiveType, PrimitiveArray};
-use arrow_buffer::{IntervalDayTime, IntervalMonthDayNano};
+use arrow_buffer::{IntervalDayTime, IntervalMonthDayNano, ScalarBuffer};
 use arrow_schema::DataType;
 use datafusion_common::DataFusionError;
 use datafusion_common::Result;
-use datafusion_physical_expr::aggregate::utils::adjust_output_array;
+
 use half::f16;
 use std::cmp::Ordering;
 use std::fmt::{Debug, Display, Formatter};
@@ -151,10 +151,11 @@ where
     }
 
     fn drain(&mut self) -> (ArrayRef, Vec<usize>) {
+        let nulls = None;
         let (vals, map_idxs) = self.heap.drain();
-        let vals = Arc::new(PrimitiveArray::<VAL>::from_iter_values(vals));
-        let vals = adjust_output_array(&self.data_type, vals).expect("Type is incorrect");
-        (vals, map_idxs)
+        let arr = PrimitiveArray::<VAL>::new(ScalarBuffer::from(vals), nulls)
+            .with_data_type(self.data_type.clone());
+        (Arc::new(arr), map_idxs)
     }
 }