You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "goutamadwant (via GitHub)" <gi...@apache.org> on 2026/08/29 04:02:03 UTC
[PR] fix: align Spark collect aggregate element nullability [datafusion]
goutamadwant opened a new pull request, #24767:
URL: https://github.com/apache/datafusion/pull/24767
## Which issue does this PR close?
- Closes #24741.
## Rationale for this change
Spark's `collect_list` and `collect_set` drop null inputs and declare arrays with `containsNull = false`. The DataFusion Spark implementations declared nullable list elements, and their accumulator results could derive nested fields from runtime arrays. The resulting schema mismatch requires downstream casts; when runtime nested fields drift from the declared type, aggregate output validation can fail.
## What changes are included in this PR?
- Declare non-nullable list elements in the return and partial-state types for both aggregates.
- Normalize partial state and final accumulator output to the declared list field, including nested child types.
- Reuse matching child arrays without casting, while reconciling a differing nested runtime type only when needed.
- Preserve the same exact type for empty and all-null results.
## Are these changes tested?
Yes. Regression tests cover return and state fields, empty results, primitive values, nested struct nullability, partial-state merging, and both `collect_list` and `collect_set`.
The existing Spark aggregate and window SQLLogicTest suites also pass. The full required workspace test suite and workspace-wide clippy with all targets and features pass.
## Are there any user-facing changes?
`collect_list` and `collect_set` now expose Arrow list schemas with non-nullable elements, matching Spark. There are no Rust API changes.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "goutamadwant (via GitHub)" <gi...@apache.org>.
goutamadwant commented on PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#issuecomment-5826800657
@sunchao addressed review comments. can you take a look when you get time and let me know if there any comments/suggestions. thanks
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "goutamadwant (via GitHub)" <gi...@apache.org>.
goutamadwant commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r4002736633
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -180,27 +234,49 @@ 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() {
+ Ok(Arc::clone(value))
+ } else {
+ Ok(cast(value.as_ref(), field.data_type())?)
+ }
+ }
}
impl<T: Accumulator> Accumulator for NullToEmptyListAccumulator<T> {
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- self.inner.update_batch(values)
+ let [value] = values else {
+ return self.inner.update_batch(values);
+ };
+ let value = self.normalize_input(value)?;
Review Comment:
Addressed with shared update/retract normalization and an update-retract-evaluate regression for the nested runtime/declared schema mismatch. Both collect_list and collect_set are covered.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "goutamadwant (via GitHub)" <gi...@apache.org>.
goutamadwant commented on PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#issuecomment-5470753487
> ## P1
> ### P1-1 - Output array column stays `nullable`, Spark declares it non-nullable
> **Where:** `SparkCollectList` / `SparkCollectSet` in `collect.rs` do not override `is_nullable()`.
>
> **Problem:** `AggregateUDFImpl::is_nullable()` defaults to `true` (`datafusion/expr/src/udaf.rs:547`), and `udaf_default_return_field` (`udaf.rs:1201`) stamps the output field's nullability from it. So both aggregates report an output field `nullable: true`. Spark's `Collect.nullable` is `false` (`collect.scala:49`); the array is never null. The PR fixed the element axis (`containsNull=false`) but left the array axis, so the Spark schema fix is half-applied.
>
> **Why it matters:** Divergence from the Spark schema this PR exists to match. It is a schema/nullability contract mismatch for the Spark-compat layer, not a wrong-value bug (the value is never null), so down-rank to P2 if the consumer (Comet) does not validate field nullability. `verdict: CONFIRMED` on the divergence; `PLAUSIBLE` on a downstream schema-check failure.
>
> **Fix:** add `fn is_nullable(&self) -> bool { false }` to both impls. Safe and precedented: `default_value` is already a non-null empty list (`empty_list_scalar`), which satisfies the `is_nullable=false` contract noted at `udaf.rs:546`. Precedent: `count.rs:296`, `approx_distinct.rs:723`, `regr.rs:467`. Assert it in `collect_types_have_non_nullable_elements` (`aggregate.is_nullable()` is not currently checked).
>
> ## P2
> ### P2-1 - No `.slt` for the schema change; existing `collect.slt` is value-only
> **Where:** `datafusion/sqllogictest/test_files/spark/aggregate/collect.slt` exists but every query is `query ?` / `query I?` (values only). The PR adds Rust unit tests, no `.slt`.
>
> **Problem:** The PR's entire behavioral change (element `containsNull=false`) is SQL-visible via `arrow_typeof`, but no SQL-level test asserts it, and the existing SLT cannot catch it (unchanged printed values). Checklist item 8 and Testing line 100: SQL-visible changes need an `.slt`; Rust unit tests bypass the planner and type plumbing.
>
> **Fix:** extend `collect.slt` with element-type assertions, e.g. `SELECT arrow_typeof(collect_list(a)) FROM (VALUES (1)) t(a);` and the `collect_set` variant. `arrow_typeof` renders the element field (`List(Field { ... nullable: false ... })`), so it pins the change. The outer `nullable=false` (P1-1) is not visible via `arrow_typeof`; cover that in the Rust test instead.
>
> ### P2-2 - `normalize_list_scalar` null-list branch is untested
> **Where:** `collect.rs`, the `array.is_null(0)` branch (new lines ~55-61, `new_null_list`).
>
> **Problem:** This branch is only reachable through `state()` on an all-null / empty group (`ArrayAggAccumulator::state()` -> `evaluate()` -> `new_null_list(..., true, 1)`, then normalized). It is real partial-aggregation behavior that flows into shuffle and `merge_batch`. `empty_results_have_non_nullable_elements` calls only `evaluate()` (which bypasses normalize via `empty_list_scalar`), and `accumulator_state_and_output_preserve_nested_type` calls `state()` only after a non-null `update_batch`. The branch has no coverage.
>
> **Fix:** add a test calling `state()` on a fresh (or all-null-updated) accumulator; assert one null list whose element field is non-nullable and named `item`, for both distinct and non-distinct.
>
> ### P2-3 - New `cast` can turn a previously-working query into a runtime error (regression risk)
> **Where:** `collect.rs`, `normalize_list_scalar` cast (new lines ~64-68): `cast(values.as_ref(), field.data_type())`.
>
> **Problem:** For nested elements (e.g. structs), when the runtime array's nested nullability is looser than the declared type (`arg_types[0]`) and the runtime data actually contains nulls under a field the declared type marks non-nullable, `cast` -> `cast_struct_to_struct` -> `StructArray::try_new` errors ("Found unmasked nulls for non-nullable field", `../arrow-rs/arrow-array/src/array/struct_array.rs:169`). Before this PR (nullable element, no normalize) the value passed through. DataFusion nullability inference is imprecise across joins, so this can regress a real query. It fails safe (errors, no corruption) and only when `values.data_type() != field.data_type()`, so primitives and matching structs are unaffected. `verdict: PLAUSIBLE`.
>
> **Fix / coverage:** add a `collect_list` test over a struct whose nested field is declared non-nullable but whose runtime input carries a null there, and decide the intended behavior (error vs. widen). `accumulator_state_and_output_preserve_nested_type` differs in nullability but uses no actual nulls, so it does not exercise this path.
Addressed in `ddbadd3`.
- Both `collect_list` and `collect_set` now report a non-nullable outer result field, with `return_field` coverage.
- `collect.slt` now pins `List(non-null Int64)` through `arrow_typeof` for both aggregates.
- Empty partial state is covered for distinct and non-distinct paths, including state merge to the final empty non-null result.
- Runtime input whose metadata is wider but whose data is valid is normalized to the declared type. An actual unmasked null under a declared non-nullable nested field is rejected fallibly at the accumulator boundary for both aggregates rather than widening the output schema or allowing `collect_set` to panic. Widening would recreate the `AggregateExec` schema mismatch this PR fixes.
Mutation checks confirmed each new test fails when its corresponding behavior is disabled. Please let me know if you ahve any other suggestions. thanks!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r4027951448
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -180,32 +234,67 @@ 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() {
+ Ok(Arc::clone(value))
+ } 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)?;
+ Ok(cast(value.as_ref(), field.data_type())?)
Review Comment:
### [P2] Remove unreachable nested payload before narrowing fields
The top-level `take` fixes the original ignored-row case, but it does not remove payload beneath null inner lists. A single retained outer list with logical value `[NULL, [1]]` still fails when the aggregate's declared element type is `List<nullable List<non-null Int32>>` and the runtime type is `List<nullable List<nullable Int32>>`.
Concrete Arrow layout:
- inner lists: offsets `[0, 1, 2]`, Int32 child `[NULL, 1]`, validity `[false, true]`;
- outer list: offsets `[0, 2]`, the inner lists as its child, and no null bitmap.
The runtime input array passes `ArrayData::validate_full()`; the only logically visible Int32 is `1`. Taking the retained outer row copies the null inner list's unused payload, so the subsequent cast rejects it with `Non-nullable field of ListArray "item" cannot contain nulls`.
I executed identical accumulator probes on `4f792b3` and merge base `c4910e0`: both `collect_list` and `collect_set` return an error from `update_batch` on the head, while both accept and return the expected logical value on the base. These are accumulator-API reproductions of the runtime/declared schema difference; I have not established a native SQL producer of that difference.
Related base-pass/head-fail cases are sliced ListView/LargeListView inputs containing only `[1]` while retaining an unused NULL backing value (both aggregates), and a dictionary of lists retaining an unused `[NULL]` dictionary value (`collect_list`).
Please account for logical reachability recursively before narrowing nested fields, including shared backing values, and add regressions for both collect functions.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "goutamadwant (via GitHub)" <gi...@apache.org>.
goutamadwant commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r4002730243
##########
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:
Confirmed in grouped and sliding-window queries. The constructor workaround does not address downstream Arrow ListArray validation. These still fail on the current branch, so this remains unresolved.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "goutamadwant (via GitHub)" <gi...@apache.org>.
goutamadwant commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r4002735149
##########
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:
Confirmed by executing dictionary scalar compaction; sparse-union compaction fails as well. Arrow's list constructors still reject these logically non-null encoded children. This remains unresolved, and the follow-up does not widen the result type or add another constructor bypass.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#issuecomment-5460323158
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/24767?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `85.71429%` with `23 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 81.51%. Comparing base ([`c4910e0`](https://app.codecov.io/gh/apache/datafusion/commit/c4910e0764f3b02cc27d347bfc117f85a4edb46d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`e3f28e6`](https://app.codecov.io/gh/apache/datafusion/commit/e3f28e6153e03b807ba636a24b75f417cd3baaa0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 8 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/24767?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [datafusion/spark/src/function/aggregate/collect.rs](https://app.codecov.io/gh/apache/datafusion/pull/24767?src=pr&el=tree&filepath=datafusion%2Fspark%2Fsrc%2Ffunction%2Faggregate%2Fcollect.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9zcGFyay9zcmMvZnVuY3Rpb24vYWdncmVnYXRlL2NvbGxlY3QucnM=) | 85.71% | [7 Missing and 16 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/24767?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #24767 +/- ##
==========================================
+ Coverage 81.47% 81.51% +0.03%
==========================================
Files 1122 1123 +1
Lines 403629 405287 +1658
Branches 403629 405287 +1658
==========================================
+ Hits 328866 330359 +1493
- Misses 55510 55584 +74
- Partials 19253 19344 +91
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/24767?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r3963773684
##########
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] Handle encoded nullability in downstream list operations
This preserves a dictionary child containing unused null entries. The resulting list passes `ArrayData::validate_full()`, but Arrow's `concat` and `take` constructors still reject its non-nullable element field.
I reproduced this with `Dictionary<Int8, Utf8>` keys `[0, 0, 0, 0]`, dictionary values `[Some("a"), None]`, group keys `[0, 0, 1, 1]`, and row IDs `[0, 1, 2, 3]`. Both queries fail on this head:
```sql
SELECT collect_list(x) FROM t GROUP BY g ORDER BY g;
SELECT collect_list(x) OVER (
ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW
) FROM t ORDER BY id;
```
The error is `Non-nullable field of ListArray "item" cannot contain nulls`. Grouped/window output concatenates list scalars through `ScalarValue::iter_to_array`, reaching the conservative constructor again. Both queries pass with merge-base production code; the corresponding `collect_set` controls pass on both versions.
Please reconcile encoded-child nullability beyond this constructor and add grouped/window regression coverage. The existing global-aggregation test produces only one scalar and misses this path.
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -180,27 +234,49 @@ 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() {
+ Ok(Arc::clone(value))
+ } else {
+ Ok(cast(value.as_ref(), field.data_type())?)
+ }
+ }
}
impl<T: Accumulator> Accumulator for NullToEmptyListAccumulator<T> {
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- self.inner.update_batch(values)
+ let [value] = values else {
+ return self.inner.update_batch(values);
+ };
+ let value = self.normalize_input(value)?;
+ self.inner.update_batch(&[value])
Review Comment:
### [P2] Normalize retractions consistently with updates
For the declared-nonnullable/runtime-nullable nested Struct scenario covered by this PR, this update path initializes `collect_set`'s `RowConverter` with the normalized schema. `retract_batch` still forwards the original runtime array.
Using declared `Struct(required: Int32 non-null)` and runtime `Struct(required: Int32 nullable)` containing `[1, 2]`, `update_batch` succeeds but retracting the first original row now fails:
```text
RowConverter column schema mismatch, expected Struct("required": non-null Int32) got Struct("required": Int32)
```
The identical update/retract probe passes on the merge base. Sliding windows pass slices of the original input to both operations, so this breaks retraction when the downstream runtime/declared schema mismatch occurs. This reproduction is at the accumulator API; I have not established a native SQL producer of that mismatch.
Please apply equivalent normalization during retraction and add an update/retract round-trip to the nested-schema regression coverage.
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -180,27 +234,49 @@ 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() {
+ Ok(Arc::clone(value))
+ } else {
+ Ok(cast(value.as_ref(), field.data_type())?)
Review Comment:
### [P2] Drop ignored null rows before narrowing nested types
This cast processes rows before the inner accumulator drops null inputs. In the runtime/declared nested-nullability mismatch scenario, it can reject the backing payload of a row that both collect aggregates must ignore.
A concrete input has declared element type `List<non-null Int32>` and runtime type `List<nullable Int32>`, with offsets `[0, 1, 2]`, child values `[NULL, 1]`, and validity `[false, true]`. Its logical rows are `[NULL, [1]]`; the null list legally retains a null backing child, and the input passes `ArrayData::validate_full()`.
Both `collect_list` and `collect_set` now fail during `update_batch` with `Non-nullable field of ListArray "item" cannot contain nulls`. The identical accumulator probes return `[[1]]` with merge-base production code. There is no null in the retained nested value.
Please remove ignored null rows before narrowing nested nullability, and cover this masked-payload case for both aggregates.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r4027954075
##########
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 at `4f792b350a6a77dfd2830dbb3833a91901862b16` with local execution and independent base/head build directories. This confirms my earlier source-traced finding: `evaluate()?.compacted()` panics in Arrow's list constructor for dictionary `collect_list` and sparse-union `collect_list`/`collect_set`. The identical three probes pass on merge base `c4910e0764f3b02cc27d347bfc117f85a4edb46d`.
I also reproduced the existing grouped/window issue using Dictionary<Int8, Utf8> keys `[0,0,1,1]`, dictionary values `[Some("a"),Some("b"),None]`, groups `[0,0,1,1]`, and IDs `[0,1,2,3]`:
```sql
SELECT g, collect_list(x) FROM t GROUP BY g ORDER BY g;
SELECT id, collect_list(x) OVER (
ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW
) FROM t ORDER BY id;
```
Both queries pass on the base and fail on the head with `Non-nullable field of ListArray "item" cannot contain nulls`, at target partitions 1 and 4. An ordered outer `array_agg` over the grouped result also fails, although that query can fail while assembling the inner grouped result; the direct compaction probes independently isolate the panic.
Dictionary `collect_set`, plain Utf8 SQL, and sliced run-end compaction controls pass on both revisions. The encoded-child issue therefore remains unresolved after the normalization/retraction follow-up.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r3963781756
##########
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] Handle encoded children during scalar compaction too
The constructor workaround preserves a dictionary child's unused null entry, so the resulting non-nullable list still fails in an ordinary consumer. With the input from `collect_list_handles_dictionary_with_unused_null` (keys `[0, 0]`, dictionary `[Some("a"), None]`), the source path for `acc.evaluate()?.compacted()` on the `collect_list` wrapper accumulator reaches [`compact_view_buffers`'s `ListArray::new`](https://github.com/apache/datafusion/blob/ddbadd312cee1dd117c4324d6e10305993fbd0dd/datafusion/common/src/scalar/mod.rs#L5011). [Arrow's dictionary copy](https://github.com/apache/arrow-rs/blob/59.2.0/arrow-data/src/transform/mod.rs#L614) retains the unused null, so that constructor sees a non-nullable field and a conservatively nullable child and panics.
This also affects composition: [ordered `array_agg` compacts each retained input scalar](https://github.com/apache/datafusion/blob/ddbadd312cee1dd117c4324d6e10305993fbd0dd/datafusion/functions-aggregate/src/array_agg.rs#L1315), and therefore fails if it receives this inner `collect_list` result. The pre-PR nullable result passed this constructor's nullability check.
Please preserve support for these valid encoded children through compaction as well, and extend the dictionary regression to compact the result or feed it into an ordered aggregate. This finding is source-traced against Arrow 59.2.0; I have not executed the proposed reproducer locally.
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -180,27 +234,49 @@ 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() {
+ Ok(Arc::clone(value))
+ } else {
+ Ok(cast(value.as_ref(), field.data_type())?)
+ }
+ }
}
impl<T: Accumulator> Accumulator for NullToEmptyListAccumulator<T> {
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- self.inner.update_batch(values)
+ let [value] = values else {
+ return self.inner.update_batch(values);
+ };
+ let value = self.normalize_input(value)?;
Review Comment:
### [P2] Normalize retracted inputs alongside updated inputs
This changes the type received by the inner accumulator, while `retract_batch` still forwards the original runtime array at lines 283–284. Reuse the valid input from `accumulator_state_and_output_preserve_nested_type`: declared `Struct(required: Int32, nullable=false)`, runtime child `nullable=true`, and actual values `[1, 2]`. After updating a `collect_set` accumulator, retracting the first row of that same runtime array reaches [`RowConverter::append`](https://github.com/apache/datafusion/blob/ddbadd312cee1dd117c4324d6e10305993fbd0dd/datafusion/functions-aggregate/src/array_agg.rs#L1110) with a different nested type. Arrow's [schema check](https://github.com/apache/arrow-rs/blob/59.2.0/arrow-row/src/lib.rs#L1069) compares nested nullability and returns `RowConverter column schema mismatch`.
Sliding windows pass slices of the original input to both operations, so the valid schema mismatch handled by the update path still causes a query error when rows leave the frame. Please apply the same normalization in `retract_batch`, and extend the nested-type test with update → retract → evaluate.
This is source-traced; I have not executed the reproducer locally or established that the complete SQL query succeeded on the pre-PR revision.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "goutamadwant (via GitHub)" <gi...@apache.org>.
goutamadwant commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r4002732432
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -180,27 +234,49 @@ 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() {
+ Ok(Arc::clone(value))
+ } else {
+ Ok(cast(value.as_ref(), field.data_type())?)
Review Comment:
Normalization now materializes only retained logical rows before narrowing nested fields, including removal of backing payload outside a slice. Both aggregates are covered through update, partial state, merge, retraction, and empty results.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "goutamadwant (via GitHub)" <gi...@apache.org>.
goutamadwant commented on code in PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#discussion_r4002731423
##########
datafusion/spark/src/function/aggregate/collect.rs:
##########
@@ -180,27 +234,49 @@ 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() {
+ Ok(Arc::clone(value))
+ } else {
+ Ok(cast(value.as_ref(), field.data_type())?)
+ }
+ }
}
impl<T: Accumulator> Accumulator for NullToEmptyListAccumulator<T> {
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- self.inner.update_batch(values)
+ let [value] = values else {
+ return self.inner.update_batch(values);
+ };
+ let value = self.normalize_input(value)?;
+ self.inner.update_batch(&[value])
Review Comment:
Retractions now use the same input normalization as updates. The regression covers nested schema differences through update, retract, state, merge, and evaluation for both aggregates.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#issuecomment-5463943261
Thanks @goutamadwant I'll check it today
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: align Spark collect aggregate element nullability [datafusion]
Posted by "goutamadwant (via GitHub)" <gi...@apache.org>.
goutamadwant commented on PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#issuecomment-5466208613
> ### [P2] Handle encoded-array nullability before rebuilding the list
> @goutamadwant The new reconstruction at [collect.rs:96–98](https://github.com/apache/datafusion/blob/e3f28e6153e03b807ba636a24b75f417cd3baaa0/datafusion/spark/src/function/aggregate/collect.rs#L96-L98) panics for valid encoded input even when every collected element is logically non-null.
>
> For example, construct a `Dictionary<Int8, Utf8>` with keys `[0, 0]` and dictionary values `[Some("a"), None]`, using its actual datatype as the aggregate input type. The array passes `ArrayData::validate_full()` and both input rows are `"a"`. Its `logical_null_count()` is zero, but Arrow 59.2's conservative `DictionaryArray::is_nullable()` is true because the dictionary retains an unused null entry.
>
> The existing logical-null filtering and single-array concatenation retain that entry. The matching-type fast path then reuses the child array, while `.with_field(field)` makes the list item non-nullable. `ListArray::new` rejects the child's `is_nullable()` and panics with:
>
> ```
> InvalidArgumentError("Non-nullable field of ListArray \"item\" cannot contain nulls")
> ```
>
> I reproduced this at `e3f28e6` through `evaluate()`, partial `state()`, and public SQL (`SELECT collect_list(x) FROM dictionary_input` after registering the batch and UDAF). The identical three probes pass with merge-base `c4910e0` production code and return `["a", "a"]`.
>
> Additional head/base controls confirm the same failure for `collect_list` over sparse unions and sliced run-end-encoded arrays, and for `collect_set` over sparse unions. The expanded matrix has 7 head panics versus all 11 cases passing on the merge base; dictionary/run-end-encoded `collect_set` controls pass, so those are not claimed as affected.
>
> Please reconcile encoded-child nullability with Arrow's construction requirements while preserving Spark's non-nullable element contract, and add regression coverage for both partial state and final output. Merely making the constructor fallible would avoid the panic but still reject these valid inputs.
@sunchao addressed this in `b8932a5d8`.
The fix now handles Arrow's conservative encoded-array nullability in `SingleRowListArrayBuilder`: for a non-nullable list field, it checks the exact logical null count and preserves the encoded child buffers and datatype when there are no logical nulls. This keeps Spark's `containsNull = false` contract and avoids the panic without decoding the values or making the field nullable.
Let me know if you have any other comments or suggestions!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org