Miscellaneous ArrayData Cleanup

datafusion/common/src/scalar.rs

@@ -2358,7 +2358,7 @@ impl ScalarValue {
                     None => v.is_null(),
                 }
             }
-            ScalarValue::Null => array.data().is_null(index),
+            ScalarValue::Null => array.is_null(index),
         }
     }
 

datafusion/core/src/avro_to_arrow/arrow_array_reader.rs

@@ -435,7 +435,7 @@ impl<'a, R: Read> AvroArrowArrayReader<'a, R> {
         });
         let valid_len = cur_offset.to_usize().unwrap();
         let array_data = match list_field.data_type() {
-            DataType::Null => NullArray::new(valid_len).data().clone(),
+            DataType::Null => NullArray::new(valid_len).into_data(),
             DataType::Boolean => {
                 let num_bytes = bit_util::ceil(valid_len, 8);
                 let mut bool_values = MutableBuffer::from_len_zeroed(num_bytes);
@@ -496,13 +496,11 @@ impl<'a, R: Read> AvroArrowArrayReader<'a, R> {
             DataType::Utf8 => flatten_string_values(rows)
                 .into_iter()
                 .collect::<StringArray>()
-                .data()
-                .clone(),
+                .into_data(),
             DataType::LargeUtf8 => flatten_string_values(rows)
                 .into_iter()
                 .collect::<LargeStringArray>()
-                .data()
-                .clone(),
+                .into_data(),
             DataType::List(field) => {
                 let child =
                     self.build_nested_list_array::<i32>(&flatten_values(rows), field)?;

datafusion/core/src/dataframe.rs

@@ -30,8 +30,8 @@ use parquet::file::properties::WriterProperties;
 use datafusion_common::from_slice::FromSlice;
 use datafusion_common::{Column, DFSchema, ScalarValue};
 use datafusion_expr::{
-    avg, count, is_null, max, median, min, stddev, TableProviderFilterPushDown,
-    UNNAMED_TABLE,
+    avg, count, is_null, max, median, min, stddev, utils::COUNT_STAR_EXPANSION,
+    TableProviderFilterPushDown, UNNAMED_TABLE,
 };
 
 use crate::arrow::datatypes::Schema;
@@ -630,7 +630,7 @@ impl DataFrame {
         let rows = self
             .aggregate(
                 vec![],
-                vec![datafusion_expr::count(Expr::Literal(ScalarValue::Null))],
+                vec![datafusion_expr::count(Expr::Literal(COUNT_STAR_EXPANSION))],
             )?
             .collect()
             .await?;

datafusion/core/src/physical_plan/file_format/parquet/row_filter.rs

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use arrow::array::{Array, BooleanArray};
+use arrow::array::BooleanArray;
 use arrow::datatypes::{DataType, Schema};
 use arrow::error::{ArrowError, Result as ArrowResult};
 use arrow::record_batch::RecordBatch;
@@ -131,8 +131,7 @@ impl ArrowPredicate for DatafusionArrowPredicate {
             .map(|v| v.into_array(batch.num_rows()))
         {
             Ok(array) => {
-                let mask = as_boolean_array(&array)?;
-                let bool_arr = BooleanArray::from(mask.data().clone());
+                let bool_arr = as_boolean_array(&array)?.clone();
                 let num_filtered = bool_arr.len() - bool_arr.true_count();
                 self.rows_filtered.add(num_filtered);
                 timer.stop();

datafusion/core/src/physical_plan/joins/utils.rs

@@ -18,11 +18,11 @@
 //! Join related functionality used both on logical and physical plans
 
 use arrow::array::{
-    new_null_array, Array, BooleanBufferBuilder, PrimitiveArray, UInt32Array,
+    downcast_array, new_null_array, Array, BooleanBufferBuilder, UInt32Array,
     UInt32Builder, UInt64Array,
 };
 use arrow::compute;
-use arrow::datatypes::{Field, Schema, UInt32Type, UInt64Type};
+use arrow::datatypes::{Field, Schema};
 use arrow::record_batch::{RecordBatch, RecordBatchOptions};
 use futures::future::{BoxFuture, Shared};
 use futures::{ready, FutureExt};
@@ -783,8 +783,8 @@ pub(crate) fn apply_join_filter_to_indices(
         filter.schema(),
         build_input_buffer,
         probe_batch,
-        PrimitiveArray::from(build_indices.data().clone()),
-        PrimitiveArray::from(probe_indices.data().clone()),
+        build_indices.clone(),
+        probe_indices.clone(),
         filter.column_indices(),
         build_side,
     )?;
@@ -794,13 +794,12 @@ pub(crate) fn apply_join_filter_to_indices(
         .into_array(intermediate_batch.num_rows());
     let mask = as_boolean_array(&filter_result)?;
 
-    let left_filtered = PrimitiveArray::<UInt64Type>::from(
-        compute::filter(&build_indices, mask)?.data().clone(),
-    );
-    let right_filtered = PrimitiveArray::<UInt32Type>::from(
-        compute::filter(&probe_indices, mask)?.data().clone(),
-    );
-    Ok((left_filtered, right_filtered))
+    let left_filtered = compute::filter(&build_indices, mask)?;
+    let right_filtered = compute::filter(&probe_indices, mask)?;
+    Ok((
+        downcast_array(left_filtered.as_ref()),
+        downcast_array(right_filtered.as_ref()),
+    ))
 }
 
 /// Returns a new [RecordBatch] by combining the `left` and `right` according to `indices`.

datafusion/physical-expr/src/aggregate/count.rs

@@ -137,13 +137,13 @@ impl Accumulator for CountAccumulator {
 
     fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
         let array = &values[0];
-        self.count += (array.len() - array.data().null_count()) as i64;
+        self.count += (array.len() - array.null_count()) as i64;
         Ok(())
     }
 
     fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
         let array = &values[0];
-        self.count -= (array.len() - array.data().null_count()) as i64;
+        self.count -= (array.len() - array.null_count()) as i64;
         Ok(())
     }
 
@@ -183,7 +183,7 @@ impl RowAccumulator for CountRowAccumulator {
         accessor: &mut RowAccessor,
     ) -> Result<()> {
         let array = &values[0];
-        let delta = (array.len() - array.data().null_count()) as u64;
+        let delta = (array.len() - array.null_count()) as u64;
         accessor.add_u64(self.state_index, delta);
         Ok(())
     }

datafusion/physical-expr/src/aggregate/sum.rs

@@ -236,15 +236,15 @@ impl Accumulator for SumAccumulator {
 
     fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
         let values = &values[0];
-        self.count += (values.len() - values.data().null_count()) as u64;
+        self.count += (values.len() - values.null_count()) as u64;
         let delta = sum_batch(values, &self.sum.get_datatype())?;
         self.sum = self.sum.add(&delta)?;
         Ok(())
     }
 
     fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
         let values = &values[0];
-        self.count -= (values.len() - values.data().null_count()) as u64;
+        self.count -= (values.len() - values.null_count()) as u64;
         let delta = sum_batch(values, &self.sum.get_datatype())?;
         self.sum = self.sum.sub(&delta)?;
         Ok(())

datafusion/physical-expr/src/expressions/case.rs

@@ -716,10 +716,10 @@ mod tests {
 
         //let valid_array = vec![true, false, false, true, false, tru
         let null_buffer = Buffer::from([0b00101001u8]);
-        let load4 = ArrayDataBuilder::new(load4.data_type().clone())
-            .len(load4.len())
+        let load4 = load4
+            .into_data()
+            .into_builder()
             .null_bit_buffer(Some(null_buffer))
-            .buffers(load4.data().buffers().to_vec())
             .build()
             .unwrap();
         let load4: Float64Array = load4.into();

datafusion/physical-expr/src/expressions/in_list.rs

@@ -86,7 +86,7 @@ where
 {
     fn new(array: &T, hash_set: ArrayHashSet) -> Self {
         Self {
-            array: T::from(array.data().clone()),
+            array: downcast_array(array),
             hash_set,
         }
     }
@@ -103,15 +103,14 @@ where
             v => {
                 let values_contains = self.contains(v.values().as_ref(), negated)?;
                 let result = take(&values_contains, v.keys(), None)?;
-                return Ok(BooleanArray::from(result.data().clone()))
+                return Ok(downcast_array(result.as_ref()))
             }
             _ => {}
         }
 
         let v = v.as_any().downcast_ref::<T>().unwrap();
-        let in_data = self.array.data();
         let in_array = &self.array;
-        let has_nulls = in_data.null_count() != 0;
+        let has_nulls = in_array.null_count() != 0;
 
         Ok(ArrayIter::new(v)
             .map(|v| {