sunchao commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r4209312252
##########
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:
### [P2] Construct a valid Map default for masked children
This new call exposes a panic in the existing
`ScalarValue::new_default(Map)` implementation: it passes Struct-typed
`ArrayData` to `MapArray::from`.
I reproduced this through `SELECT collect_list(x) FROM nested_input` after
registering a fully Arrow-validated batch with matching declared/runtime types.
The single outer struct is valid, its nullable `optional` struct is null, and
that inner struct contains a required Map<Int32, Int32> child whose backing
entry is `{1: 7}`. None of that map payload is logically visible, but
normalization now tries to construct its default and panics:
```text
MapArray expected ArrayData with DataType::Map got Struct("key": non-null
Int32, "value": non-null Int32)
```
The identical SQL probe succeeds on base `696eaf5` and preserves the null
nested struct. An integer-only sparse union with an inactive Map variant also
panics through the new default call at lines 653–654.
Please construct a valid, correctly typed replacement here or repair the Map
default helper, and cover masked/inactive Map children. The helper defect
predates this PR; making it reachable from these valid collect inputs is the
regression.
##########
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:
### [P2] Avoid copying real payload into every masked row
Using the first valid value as the `take` fallback copies its entire
variable-length payload into every null parent row. This introduces a large
memory regression for ordinary SQL, even when runtime and declared types
already match.
I registered 1,024 rows with `id = 0..1023` and a Utf8 `payload` column
containing one 4,096-byte string at id 0 and NULL elsewhere, then ran:
```sql
SELECT collect_list(
named_struct('optional',
CASE WHEN id = 0 THEN named_struct('payload', payload) ELSE NULL END)
)
FROM struct_source;
```
With target partitions 1 and batch size 1,024, the projected input retains
4,096 string bytes. Base `696eaf5` retains 4,096 bytes in the collected result;
head `b259eba` retains 4,194,304 bytes. Every logical value is unchanged: one
populated inner struct and 1,023 null inner structs. These are measured
retained Utf8 payload bytes, not process RSS or query timing.
Sparse-union filling at lines 667–675 has the same amplification when one
active string is copied into inactive slots. Please use inexpensive, correctly
typed masked values or preserve safe masked storage instead of duplicating real
payload.
##########
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:
### [P2] Preserve the length when rebuilding zero-field structs
`StructArray::try_new` cannot infer a length when `fields` and `columns` are
empty. The new recursive path reaches it for a valid nullable empty struct.
I registered one fully Arrow-validated, exact-schema outer struct whose
nullable `optional` child is `StructArray::new_empty_fields(1,
Some(NullBuffer::new_null(1)))`, then executed `SELECT collect_list(x) FROM
nested_input`. Base `696eaf5` succeeds and preserves the null child; head
`b259eba` returns:
```text
use StructArray::try_new_with_length or StructArray::new_empty_fields to
create a struct array with no fields so that the length can be set correctly
```
Please reconstruct with `StructArray::try_new_with_length(...,
source.len())` and add this zero-field nested case.
##########
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:
### [P2] Preserve shared dense-union offsets before taking rows
This unconditional `take` expands shared dense-union child offsets before
`normalize_dense_union` can compact them. Arrow takes each referenced child
occurrence separately and rewrites the offsets, so the later `BTreeSet` sees
distinct offsets and cannot recover the sharing.
A valid exact-type input with 1,024 dense-union rows, all using the string
variant and offset 0, needs only one 4,096-byte Utf8 child value. Through
`collect_list` update/evaluate, base `696eaf5` retains that one value (4,096
bytes), while head `b259eba` retains 1,024 copies (4,194,304 bytes). The probe
validates Arrow data and checks every logical result.
Please normalize/remap the union references without duplicating each child
occurrence—for example, bypass the top-level take when no logical nulls need
removing and preserve shared offsets when filtering. This is a separate path
from the masked-row fallback: replacing those fillers alone will not fix this
expansion.
##########
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:
### [P2] Inactive Null union variants still fail during result assembly
Rechecked at `b259ebacde45a6e61362ca87747836ae33068538`. The earlier
dictionary grouped/window cases and direct compaction reproducers now pass, but
this downstream encoded-child issue remains for a sparse Union<Int32, Null>.
Concrete validated fixture: fields `integer: Int32 non-null` and `null: Null
nullable`, type IDs `[0,0,0,0]`, integer values `[1,2,1,2]`, an inactive
`NullArray(4)`, groups `[0,0,1,1]`, and IDs `[0,1,2,3]`. Every logical input
value is a non-null integer. With target partitions 1, both queries fail on the
head:
```sql
SELECT g, collect_list(x) FROM sparse_null_arm GROUP BY g ORDER BY g;
SELECT id, collect_list(x) OVER (
ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW
) FROM sparse_null_arm ORDER BY id;
```
Replacing `collect_list` with `collect_set` also fails. All four identical
SQL probes pass on base `696eaf58d436faa6532aa0f5875b5a120b94d0eb`. The error
remains `Non-nullable field of ListArray "item" cannot contain nulls`.
The inactive-variant branch in
[`normalize_sparse_union`](https://github.com/apache/datafusion/blob/b259ebacde45a6e61362ca87747836ae33068538/datafusion/spark/src/function/aggregate/collect.rs#L653-L654)
produces another NullArray, so the union remains conservatively nullable
despite having zero logical nulls. The builder/compaction bypasses accept it,
but Arrow's list constructor rejects it when multiple aggregate results are
assembled. Please cover this case in grouped/window output handling; scrubbing
unused payload cannot make a NullArray non-nullable.
--
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]
