goutamadwant commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r4521380471
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -43,7 +51,811 @@ fn empty_list_scalar(list_type: &DataType) ->
Result<ScalarValue> {
);
};
let empty = arrow::array::new_empty_array(field.data_type());
- Ok(SingleRowListArrayBuilder::new(empty).build_list_scalar())
+ Ok(SingleRowListArrayBuilder::new(empty)
+ .with_field(field)
+ .build_list_scalar())
+}
+
+fn collect_type(element_type: DataType) -> DataType {
+ DataType::List(Arc::new(Field::new_list_field(element_type, false)))
+}
+
+/// Whether an exact-type array still needs normalization to discard physical
+/// storage that is not part of its logical value.
+fn has_compactable_storage(value: &ArrayRef) -> bool {
+ match value.data_type() {
+ DataType::Dictionary(_, _)
+ | DataType::RunEndEncoded(_, _)
+ | DataType::ListView(_)
+ | DataType::LargeListView(_)
+ | DataType::Union(_, _) => true,
+ DataType::List(_) => {
+ let list = value.as_list::<i32>();
+ list.null_count() != 0
+ || list
+ .value_offsets()
+ .first()
+ .is_some_and(|offset| *offset != 0)
+ || list
+ .value_offsets()
+ .last()
+ .is_some_and(|offset| *offset as usize !=
list.values().len())
+ || has_compactable_storage(list.values())
+ }
+ DataType::LargeList(_) => {
+ let list = value.as_list::<i64>();
+ list.null_count() != 0
+ || list
+ .value_offsets()
+ .first()
+ .is_some_and(|offset| *offset != 0)
+ || list
+ .value_offsets()
+ .last()
+ .is_some_and(|offset| *offset as usize !=
list.values().len())
+ || has_compactable_storage(list.values())
+ }
+ DataType::FixedSizeList(_, size) => {
+ let list = value.as_fixed_size_list();
+ let has_exact_child_length = usize::try_from(*size)
+ .ok()
+ .and_then(|size| list.len().checked_mul(size))
+ .is_some_and(|length| length == list.values().len());
+ list.null_count() != 0
+ || list.values().offset() != 0
+ || !has_exact_child_length
+ || has_compactable_storage(list.values())
+ }
+ DataType::Struct(_) => {
+ let structure = value.as_struct();
+ structure.null_count() != 0
+ || structure.columns().iter().any(has_compactable_storage)
+ }
+ DataType::Map(_, _) => {
+ let map = value.as_map();
+ map.null_count() != 0
+ || map
+ .value_offsets()
+ .first()
+ .is_some_and(|offset| *offset != 0)
+ || map
+ .value_offsets()
+ .last()
+ .is_some_and(|offset| *offset as usize !=
map.entries().len())
+ || map.entries().columns().iter().any(has_compactable_storage)
+ }
+ _ => false,
+ }
+}
+
+/// Materialize only logically reachable values and align their nested type
with
+/// `target_type`.
+///
+/// Arrow container arrays may retain unreachable null payload in null lists,
+/// unused dictionary values, sparse-union children, and view backing buffers.
+/// That payload is valid, but it makes `Array::is_nullable` conservative and
+/// prevents the result from being embedded below Spark's non-null list field.
+fn normalize_array(
+ value: &ArrayRef,
+ target_type: &DataType,
+ require_non_null: bool,
+) -> Result<ArrayRef> {
+ if value.data_type() == target_type
+ && !has_compactable_storage(value)
+ && (!require_non_null || !value.is_nullable())
+ {
+ return Ok(Arc::clone(value));
+ }
+ let normalized = match (value.data_type(), target_type) {
+ (DataType::List(_), DataType::List(field)) => {
+ normalize_list::<i32>(value, field)?
+ }
+ (DataType::LargeList(_), DataType::LargeList(field)) => {
+ normalize_list::<i64>(value, field)?
+ }
+ (DataType::ListView(_), DataType::ListView(field)) => {
+ normalize_list_view::<i32>(value, field)?
+ }
+ (DataType::LargeListView(_), DataType::LargeListView(field)) => {
+ normalize_list_view::<i64>(value, field)?
+ }
+ (
+ DataType::FixedSizeList(_, source_size),
+ DataType::FixedSizeList(field, target_size),
+ ) if source_size == target_size => {
+ normalize_fixed_size_list(value, field, *target_size)?
+ }
+ (
+ DataType::Dictionary(source_key, _),
+ DataType::Dictionary(target_key, target_value),
+ ) if source_key == target_key => {
+ let compact =
garbage_collect_any_dictionary(value.as_any_dictionary())?;
+ let dictionary = compact.as_any_dictionary();
+ let values =
+ normalize_array(dictionary.values(), target_value,
require_non_null)?;
+ let dictionary = dictionary.with_values(values);
+ if dictionary.null_count() == 0 &&
dictionary.to_data().nulls().is_some() {
+ let data =
dictionary.to_data().into_builder().nulls(None).build()?;
+ make_array(data)
+ } else {
+ dictionary
+ }
+ }
+ (
+ DataType::RunEndEncoded(source_run_ends, _),
+ DataType::RunEndEncoded(target_run_ends, target_value),
+ ) if source_run_ends.data_type() == target_run_ends.data_type() => {
+ // Work at the physical run count, not the potentially enormous
+ // logical length represented by those runs.
+ let value = value.as_ref();
+ arrow::array::downcast_run_array! {
+ value => normalize_run_end(
+ value,
+ target_type,
+ target_value.data_type(),
+ require_non_null,
+ ),
+ _ => internal_err!("collect_list/collect_set expected a
run-end encoded array"),
+ }?
+ }
+ (DataType::Struct(_), DataType::Struct(fields)) => {
+ let source = value.as_struct();
+ let columns = fields
+ .iter()
+ .zip(source.columns())
+ .map(|(field, column)| {
+ let column = materialize_masked_values(
+ column,
+ source.nulls(),
+ field.data_type(),
+ !field.is_nullable(),
+ )?;
+ normalize_array(&column, field.data_type(),
!field.is_nullable())
+ })
+ .collect::<Result<Vec<_>>>()?;
+ Arc::new(StructArray::try_new(
+ fields.clone(),
+ columns,
+ source.nulls().cloned(),
+ )?) as ArrayRef
+ }
+ (DataType::Map(_, _), DataType::Map(field, ordered)) => {
+ normalize_map(value, field, *ordered)?
+ }
+ (
+ DataType::Union(source_fields, UnionMode::Sparse),
+ DataType::Union(fields, UnionMode::Sparse),
+ ) => normalize_sparse_union(value, source_fields, fields,
require_non_null)?,
+ (
+ DataType::Union(source_fields, UnionMode::Dense),
+ DataType::Union(fields, UnionMode::Dense),
+ ) => normalize_dense_union(value, source_fields, fields,
require_non_null)?,
+ _ => {
+ let value = if value.data_type() == target_type {
+ Arc::clone(value)
+ } else {
+ cast(value.as_ref(), target_type)?
+ };
+ if require_non_null
+ && value.logical_null_count() == 0
+ && value.is_nullable()
+ && value.to_data().nulls().is_some()
+ {
+ let data = value.to_data().into_builder().nulls(None).build()?;
+ make_array(data)
+ } else {
+ value
+ }
+ }
+ };
+ Ok(normalized)
+}
+
+fn materialize_masked_values(
+ value: &ArrayRef,
+ parent_nulls: Option<&arrow::buffer::NullBuffer>,
+ target_type: &DataType,
+ require_non_null: bool,
+) -> Result<ArrayRef> {
+ let Some(parent_nulls) = parent_nulls else {
+ return Ok(Arc::clone(value));
+ };
+ let mut valid_parent_indices = parent_nulls.valid_indices();
+ let Some(first_valid_parent) = valid_parent_indices.next() else {
+ return
ScalarValue::new_default(target_type)?.to_array_of_size(value.len());
Review Comment:
Fixed the Map default to preserve its type and sortedness, with tests for
masked and inactive Map children.
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -43,7 +51,811 @@ fn empty_list_scalar(list_type: &DataType) ->
Result<ScalarValue> {
);
};
let empty = arrow::array::new_empty_array(field.data_type());
- Ok(SingleRowListArrayBuilder::new(empty).build_list_scalar())
+ Ok(SingleRowListArrayBuilder::new(empty)
+ .with_field(field)
+ .build_list_scalar())
+}
+
+fn collect_type(element_type: DataType) -> DataType {
+ DataType::List(Arc::new(Field::new_list_field(element_type, false)))
+}
+
+/// Whether an exact-type array still needs normalization to discard physical
+/// storage that is not part of its logical value.
+fn has_compactable_storage(value: &ArrayRef) -> bool {
+ match value.data_type() {
+ DataType::Dictionary(_, _)
+ | DataType::RunEndEncoded(_, _)
+ | DataType::ListView(_)
+ | DataType::LargeListView(_)
+ | DataType::Union(_, _) => true,
+ DataType::List(_) => {
+ let list = value.as_list::<i32>();
+ list.null_count() != 0
+ || list
+ .value_offsets()
+ .first()
+ .is_some_and(|offset| *offset != 0)
+ || list
+ .value_offsets()
+ .last()
+ .is_some_and(|offset| *offset as usize !=
list.values().len())
+ || has_compactable_storage(list.values())
+ }
+ DataType::LargeList(_) => {
+ let list = value.as_list::<i64>();
+ list.null_count() != 0
+ || list
+ .value_offsets()
+ .first()
+ .is_some_and(|offset| *offset != 0)
+ || list
+ .value_offsets()
+ .last()
+ .is_some_and(|offset| *offset as usize !=
list.values().len())
+ || has_compactable_storage(list.values())
+ }
+ DataType::FixedSizeList(_, size) => {
+ let list = value.as_fixed_size_list();
+ let has_exact_child_length = usize::try_from(*size)
+ .ok()
+ .and_then(|size| list.len().checked_mul(size))
+ .is_some_and(|length| length == list.values().len());
+ list.null_count() != 0
+ || list.values().offset() != 0
+ || !has_exact_child_length
+ || has_compactable_storage(list.values())
+ }
+ DataType::Struct(_) => {
+ let structure = value.as_struct();
+ structure.null_count() != 0
+ || structure.columns().iter().any(has_compactable_storage)
+ }
+ DataType::Map(_, _) => {
+ let map = value.as_map();
+ map.null_count() != 0
+ || map
+ .value_offsets()
+ .first()
+ .is_some_and(|offset| *offset != 0)
+ || map
+ .value_offsets()
+ .last()
+ .is_some_and(|offset| *offset as usize !=
map.entries().len())
+ || map.entries().columns().iter().any(has_compactable_storage)
+ }
+ _ => false,
+ }
+}
+
+/// Materialize only logically reachable values and align their nested type
with
+/// `target_type`.
+///
+/// Arrow container arrays may retain unreachable null payload in null lists,
+/// unused dictionary values, sparse-union children, and view backing buffers.
+/// That payload is valid, but it makes `Array::is_nullable` conservative and
+/// prevents the result from being embedded below Spark's non-null list field.
+fn normalize_array(
+ value: &ArrayRef,
+ target_type: &DataType,
+ require_non_null: bool,
+) -> Result<ArrayRef> {
+ if value.data_type() == target_type
+ && !has_compactable_storage(value)
+ && (!require_non_null || !value.is_nullable())
+ {
+ return Ok(Arc::clone(value));
+ }
+ let normalized = match (value.data_type(), target_type) {
+ (DataType::List(_), DataType::List(field)) => {
+ normalize_list::<i32>(value, field)?
+ }
+ (DataType::LargeList(_), DataType::LargeList(field)) => {
+ normalize_list::<i64>(value, field)?
+ }
+ (DataType::ListView(_), DataType::ListView(field)) => {
+ normalize_list_view::<i32>(value, field)?
+ }
+ (DataType::LargeListView(_), DataType::LargeListView(field)) => {
+ normalize_list_view::<i64>(value, field)?
+ }
+ (
+ DataType::FixedSizeList(_, source_size),
+ DataType::FixedSizeList(field, target_size),
+ ) if source_size == target_size => {
+ normalize_fixed_size_list(value, field, *target_size)?
+ }
+ (
+ DataType::Dictionary(source_key, _),
+ DataType::Dictionary(target_key, target_value),
+ ) if source_key == target_key => {
+ let compact =
garbage_collect_any_dictionary(value.as_any_dictionary())?;
+ let dictionary = compact.as_any_dictionary();
+ let values =
+ normalize_array(dictionary.values(), target_value,
require_non_null)?;
+ let dictionary = dictionary.with_values(values);
+ if dictionary.null_count() == 0 &&
dictionary.to_data().nulls().is_some() {
+ let data =
dictionary.to_data().into_builder().nulls(None).build()?;
+ make_array(data)
+ } else {
+ dictionary
+ }
+ }
+ (
+ DataType::RunEndEncoded(source_run_ends, _),
+ DataType::RunEndEncoded(target_run_ends, target_value),
+ ) if source_run_ends.data_type() == target_run_ends.data_type() => {
+ // Work at the physical run count, not the potentially enormous
+ // logical length represented by those runs.
+ let value = value.as_ref();
+ arrow::array::downcast_run_array! {
+ value => normalize_run_end(
+ value,
+ target_type,
+ target_value.data_type(),
+ require_non_null,
+ ),
+ _ => internal_err!("collect_list/collect_set expected a
run-end encoded array"),
+ }?
+ }
+ (DataType::Struct(_), DataType::Struct(fields)) => {
+ let source = value.as_struct();
+ let columns = fields
+ .iter()
+ .zip(source.columns())
+ .map(|(field, column)| {
+ let column = materialize_masked_values(
+ column,
+ source.nulls(),
+ field.data_type(),
+ !field.is_nullable(),
+ )?;
+ normalize_array(&column, field.data_type(),
!field.is_nullable())
+ })
+ .collect::<Result<Vec<_>>>()?;
+ Arc::new(StructArray::try_new(
+ fields.clone(),
+ columns,
+ source.nulls().cloned(),
+ )?) as ArrayRef
+ }
+ (DataType::Map(_, _), DataType::Map(field, ordered)) => {
+ normalize_map(value, field, *ordered)?
+ }
+ (
+ DataType::Union(source_fields, UnionMode::Sparse),
+ DataType::Union(fields, UnionMode::Sparse),
+ ) => normalize_sparse_union(value, source_fields, fields,
require_non_null)?,
+ (
+ DataType::Union(source_fields, UnionMode::Dense),
+ DataType::Union(fields, UnionMode::Dense),
+ ) => normalize_dense_union(value, source_fields, fields,
require_non_null)?,
+ _ => {
+ let value = if value.data_type() == target_type {
+ Arc::clone(value)
+ } else {
+ cast(value.as_ref(), target_type)?
+ };
+ if require_non_null
+ && value.logical_null_count() == 0
+ && value.is_nullable()
+ && value.to_data().nulls().is_some()
+ {
+ let data = value.to_data().into_builder().nulls(None).build()?;
+ make_array(data)
+ } else {
+ value
+ }
+ }
+ };
+ Ok(normalized)
+}
+
+fn materialize_masked_values(
+ value: &ArrayRef,
+ parent_nulls: Option<&arrow::buffer::NullBuffer>,
+ target_type: &DataType,
+ require_non_null: bool,
+) -> Result<ArrayRef> {
+ let Some(parent_nulls) = parent_nulls else {
+ return Ok(Arc::clone(value));
+ };
+ let mut valid_parent_indices = parent_nulls.valid_indices();
+ let Some(first_valid_parent) = valid_parent_indices.next() else {
+ return
ScalarValue::new_default(target_type)?.to_array_of_size(value.len());
+ };
+ let fallback = if require_non_null {
+ std::iter::once(first_valid_parent)
+ .chain(valid_parent_indices)
+ .find(|index| value.is_valid(*index))
+ } else {
+ Some(first_valid_parent)
+ };
+ let Some(fallback) = fallback else {
+ // The nulls are logically visible because the parent rows are valid.
+ // Preserve them so the declared non-null field validation rejects the
+ // input instead of silently replacing user data with defaults.
+ return Ok(Arc::clone(value));
+ };
+ let indices = UInt64Array::from_iter_values((0..value.len()).map(|index| {
+ if parent_nulls.is_valid(index) {
+ index as u64
+ } else {
+ fallback as u64
+ }
+ }));
+ Ok(take(value.as_ref(), &indices, None)?)
Review Comment:
Masked slots now use cheap typed defaults instead of copying visible
payload. Added SQL and sparse union regressions for the memory growth.
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -180,32 +994,78 @@ impl<T: Accumulator> NullToEmptyListAccumulator<T> {
pub fn new(inner: T, list_type: DataType) -> Self {
Self { inner, list_type }
}
+
+ fn normalize_input(&self, value: &ArrayRef) -> Result<ArrayRef> {
+ let DataType::List(field) = &self.list_type else {
+ return internal_err!(
+ "collect_list/collect_set expected List return type, got {:?}",
+ self.list_type
+ );
+ };
+ if value.data_type() == field.data_type() &&
!has_compactable_storage(value) {
+ Ok(Arc::clone(value))
+ } else if matches!(
+ (value.data_type(), field.data_type()),
+ (
+ DataType::RunEndEncoded(source_run_ends, _),
+ DataType::RunEndEncoded(target_run_ends, _)
+ ) if source_run_ends.data_type() == target_run_ends.data_type()
+ ) {
+ // Normalizing a run-end encoded array through `take` requires an
+ // index for every logical row. Dispatch it directly so work stays
+ // bounded by the physical run count.
+ normalize_array(value, field.data_type(), false)
+ } else {
+ // Materialize only retained rows before narrowing nested fields.
+ // A slice can still reference null payload outside its logical
rows.
+ let indices = match value.logical_nulls() {
+ Some(nulls) => UInt64Array::from_iter_values(
+ nulls.valid_indices().map(|index| index as u64),
+ ),
+ None => UInt64Array::from_iter_values(0..value.len() as u64),
+ };
+ let value = take(value.as_ref(), &indices, None)?;
Review Comment:
Row selection now preserves shared dense union offsets, with tests checking
values and retained backing size across the nested paths we covered.
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -43,7 +51,811 @@ fn empty_list_scalar(list_type: &DataType) ->
Result<ScalarValue> {
);
};
let empty = arrow::array::new_empty_array(field.data_type());
- Ok(SingleRowListArrayBuilder::new(empty).build_list_scalar())
+ Ok(SingleRowListArrayBuilder::new(empty)
+ .with_field(field)
+ .build_list_scalar())
+}
+
+fn collect_type(element_type: DataType) -> DataType {
+ DataType::List(Arc::new(Field::new_list_field(element_type, false)))
+}
+
+/// Whether an exact-type array still needs normalization to discard physical
+/// storage that is not part of its logical value.
+fn has_compactable_storage(value: &ArrayRef) -> bool {
+ match value.data_type() {
+ DataType::Dictionary(_, _)
+ | DataType::RunEndEncoded(_, _)
+ | DataType::ListView(_)
+ | DataType::LargeListView(_)
+ | DataType::Union(_, _) => true,
+ DataType::List(_) => {
+ let list = value.as_list::<i32>();
+ list.null_count() != 0
+ || list
+ .value_offsets()
+ .first()
+ .is_some_and(|offset| *offset != 0)
+ || list
+ .value_offsets()
+ .last()
+ .is_some_and(|offset| *offset as usize !=
list.values().len())
+ || has_compactable_storage(list.values())
+ }
+ DataType::LargeList(_) => {
+ let list = value.as_list::<i64>();
+ list.null_count() != 0
+ || list
+ .value_offsets()
+ .first()
+ .is_some_and(|offset| *offset != 0)
+ || list
+ .value_offsets()
+ .last()
+ .is_some_and(|offset| *offset as usize !=
list.values().len())
+ || has_compactable_storage(list.values())
+ }
+ DataType::FixedSizeList(_, size) => {
+ let list = value.as_fixed_size_list();
+ let has_exact_child_length = usize::try_from(*size)
+ .ok()
+ .and_then(|size| list.len().checked_mul(size))
+ .is_some_and(|length| length == list.values().len());
+ list.null_count() != 0
+ || list.values().offset() != 0
+ || !has_exact_child_length
+ || has_compactable_storage(list.values())
+ }
+ DataType::Struct(_) => {
+ let structure = value.as_struct();
+ structure.null_count() != 0
+ || structure.columns().iter().any(has_compactable_storage)
+ }
+ DataType::Map(_, _) => {
+ let map = value.as_map();
+ map.null_count() != 0
+ || map
+ .value_offsets()
+ .first()
+ .is_some_and(|offset| *offset != 0)
+ || map
+ .value_offsets()
+ .last()
+ .is_some_and(|offset| *offset as usize !=
map.entries().len())
+ || map.entries().columns().iter().any(has_compactable_storage)
+ }
+ _ => false,
+ }
+}
+
+/// Materialize only logically reachable values and align their nested type
with
+/// `target_type`.
+///
+/// Arrow container arrays may retain unreachable null payload in null lists,
+/// unused dictionary values, sparse-union children, and view backing buffers.
+/// That payload is valid, but it makes `Array::is_nullable` conservative and
+/// prevents the result from being embedded below Spark's non-null list field.
+fn normalize_array(
+ value: &ArrayRef,
+ target_type: &DataType,
+ require_non_null: bool,
+) -> Result<ArrayRef> {
+ if value.data_type() == target_type
+ && !has_compactable_storage(value)
+ && (!require_non_null || !value.is_nullable())
+ {
+ return Ok(Arc::clone(value));
+ }
+ let normalized = match (value.data_type(), target_type) {
+ (DataType::List(_), DataType::List(field)) => {
+ normalize_list::<i32>(value, field)?
+ }
+ (DataType::LargeList(_), DataType::LargeList(field)) => {
+ normalize_list::<i64>(value, field)?
+ }
+ (DataType::ListView(_), DataType::ListView(field)) => {
+ normalize_list_view::<i32>(value, field)?
+ }
+ (DataType::LargeListView(_), DataType::LargeListView(field)) => {
+ normalize_list_view::<i64>(value, field)?
+ }
+ (
+ DataType::FixedSizeList(_, source_size),
+ DataType::FixedSizeList(field, target_size),
+ ) if source_size == target_size => {
+ normalize_fixed_size_list(value, field, *target_size)?
+ }
+ (
+ DataType::Dictionary(source_key, _),
+ DataType::Dictionary(target_key, target_value),
+ ) if source_key == target_key => {
+ let compact =
garbage_collect_any_dictionary(value.as_any_dictionary())?;
+ let dictionary = compact.as_any_dictionary();
+ let values =
+ normalize_array(dictionary.values(), target_value,
require_non_null)?;
+ let dictionary = dictionary.with_values(values);
+ if dictionary.null_count() == 0 &&
dictionary.to_data().nulls().is_some() {
+ let data =
dictionary.to_data().into_builder().nulls(None).build()?;
+ make_array(data)
+ } else {
+ dictionary
+ }
+ }
+ (
+ DataType::RunEndEncoded(source_run_ends, _),
+ DataType::RunEndEncoded(target_run_ends, target_value),
+ ) if source_run_ends.data_type() == target_run_ends.data_type() => {
+ // Work at the physical run count, not the potentially enormous
+ // logical length represented by those runs.
+ let value = value.as_ref();
+ arrow::array::downcast_run_array! {
+ value => normalize_run_end(
+ value,
+ target_type,
+ target_value.data_type(),
+ require_non_null,
+ ),
+ _ => internal_err!("collect_list/collect_set expected a
run-end encoded array"),
+ }?
+ }
+ (DataType::Struct(_), DataType::Struct(fields)) => {
+ let source = value.as_struct();
+ let columns = fields
+ .iter()
+ .zip(source.columns())
+ .map(|(field, column)| {
+ let column = materialize_masked_values(
+ column,
+ source.nulls(),
+ field.data_type(),
+ !field.is_nullable(),
+ )?;
+ normalize_array(&column, field.data_type(),
!field.is_nullable())
+ })
+ .collect::<Result<Vec<_>>>()?;
+ Arc::new(StructArray::try_new(
+ fields.clone(),
+ columns,
+ source.nulls().cloned(),
+ )?) as ArrayRef
Review Comment:
Fixed this by preserving the struct length explicitly and added a regression
for the nullable empty struct case.
##########
datafusion/common/src/utils/mod.rs:
##########
@@ -603,9 +601,31 @@ impl SingleRowListArrayBuilder {
/// Build a single element [`LargeListArray`]
pub fn build_large_list_array(self) -> LargeListArray {
+ self.build_generic_list_array()
+ }
+
+ fn build_generic_list_array<OffsetSize: OffsetSizeTrait>(
+ self,
+ ) -> GenericListArray<OffsetSize> {
let (field, arr) = self.into_field_and_arr();
let offsets = OffsetBuffer::from_lengths([arr.len()]);
- LargeListArray::new(field, offsets, arr, None)
+
+ // `is_nullable` is conservative for encoded arrays and can be true
+ // even when the array contains no logical nulls. In that case the
+ // generic constructor rejects a valid non-nullable list child.
+ if !field.is_nullable() && arr.is_nullable() &&
arr.logical_null_count() == 0 {
+ let data = ArrayData::builder(
+ GenericListArray::<OffsetSize>::DATA_TYPE_CONSTRUCTOR(field),
+ )
+ .len(1)
+ .add_buffer(offsets.into_inner().into_inner())
+ .add_child_data(arr.to_data())
+ .build()
+ .expect("single-row list array should contain valid data");
+ return GenericListArray::from(data);
Review Comment:
Rechecked with Arrow 60.1 from main. Grouped queries, sliding windows and
compaction now pass for both aggregates with 1 and 4 partitions.
##########
datafusion/common/src/utils/mod.rs:
##########
@@ -603,9 +601,31 @@ impl SingleRowListArrayBuilder {
/// Build a single element [`LargeListArray`]
pub fn build_large_list_array(self) -> LargeListArray {
+ self.build_generic_list_array()
+ }
+
+ fn build_generic_list_array<OffsetSize: OffsetSizeTrait>(
+ self,
+ ) -> GenericListArray<OffsetSize> {
let (field, arr) = self.into_field_and_arr();
let offsets = OffsetBuffer::from_lengths([arr.len()]);
- LargeListArray::new(field, offsets, arr, None)
+
+ // `is_nullable` is conservative for encoded arrays and can be true
+ // even when the array contains no logical nulls. In that case the
+ // generic constructor rejects a valid non-nullable list child.
+ if !field.is_nullable() && arr.is_nullable() &&
arr.logical_null_count() == 0 {
+ let data = ArrayData::builder(
+ GenericListArray::<OffsetSize>::DATA_TYPE_CONSTRUCTOR(field),
+ )
+ .len(1)
+ .add_buffer(offsets.into_inner().into_inner())
+ .add_child_data(arr.to_data())
+ .build()
+ .expect("single-row list array should contain valid data");
+ return GenericListArray::from(data);
Review Comment:
Rechecked on Arrow 60.1. Dictionary and union results now pass compaction
and grouped/window queries without the constructor workarounds.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
