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: [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]
