Skip to content

fix: align Spark collect aggregate element nullability #24767

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
goutamadwant wants to merge 14 commits into
base: main
Choose a base branch
from
Open

fix: align Spark collect aggregate element nullability #24767

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
e3f28e6
fix: align Spark collect aggregate element nullability
goutamadwant Aug 28, 2026
ddbadd3
fix: handle encoded values in non-nullable list builder
goutamadwant Aug 30, 2026
4f792b3
fix: normalize Spark collect retractions and retained input rows
goutamadwant Sep 14, 2026
735a4b3
fix: normalize Spark collect encoded values
goutamadwant Sep 19, 2026
d7d26bd
fix: bound collect encoded normalization
goutamadwant Sep 19, 2026
eb0a76d
fix: harden collect encoded normalization
goutamadwant Sep 19, 2026
3cf0870
fix: handle inactive sparse union variants
goutamadwant Sep 19, 2026
cdba3e3
fix: preserve nullable run values during collect normalization
goutamadwant Sep 19, 2026
985de98
fix: preserve nullable dictionary values during collect normalization
goutamadwant Sep 19, 2026
ae8c9c0
fix: compact exact encoded collect values
goutamadwant Sep 19, 2026
bfa35cc
fix: compact nested collect wrappers
goutamadwant Sep 19, 2026
209b774
Merge remote-tracking branch 'origin/main' into fix-spark-collect-nul…
goutamadwant Sep 19, 2026
d65cf5f
fix: preserve fixed-size-list default shape
goutamadwant Sep 19, 2026
f83f0f4
fix: harden Spark collect normalization
goutamadwant Sep 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

55 changes: 41 additions & 14 deletions datafusion/common/src/scalar/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1724,14 +1724,21 @@ impl ScalarValue {
ScalarValue::new_list(&[], field.data_type(), field.is_nullable());
Ok(ScalarValue::List(list))
}
DataType::FixedSizeList(field, _size) => {
let empty_arr = new_empty_array(field.data_type());
let values = Arc::new(
SingleRowListArrayBuilder::new(empty_arr)
.with_field(field)
.build_fixed_size_list_array(0),
);
Ok(ScalarValue::FixedSizeList(values))
DataType::FixedSizeList(field, size) => {
let list_size = size.to_usize().ok_or_else(|| {
_internal_datafusion_err!("FixedSizeList size cannot be negative")
})?;
let values = ScalarValue::new_default(field.data_type())?
.to_array_of_size(list_size)?;
Ok(ScalarValue::FixedSizeList(Arc::new(
FixedSizeListArray::try_new_with_length(
Arc::clone(field),
*size,
values,
None,
1,
)?,
)))
}
DataType::LargeList(field) => {
let list = ScalarValue::new_large_list(&[], field.data_type());
Expand Down Expand Up @@ -5038,12 +5045,32 @@ impl ScalarValue {
macro_rules! gc_list {
($field:expr, $offset_type:ty, $array_type:ty) => {{
let list = array.as_list::<$offset_type>();
Arc::new(<$array_type>::new(
Arc::clone($field),
list.offsets().clone(),
ScalarValue::compact_view_buffers(Arc::clone(list.values())),
list.nulls().cloned(),
)) as ArrayRef
let offsets = list.offsets().clone();
let values = ScalarValue::compact_view_buffers(Arc::clone(list.values()));
if !$field.is_nullable()
&& values.is_nullable()
&& values.logical_null_count() == 0
{
let data = ArrayData::builder(
GenericListArray::<$offset_type>::DATA_TYPE_CONSTRUCTOR(
Arc::clone($field),
),
)
.len(list.len())
.nulls(list.nulls().cloned())
.add_buffer(offsets.into_inner().into_inner())
.add_child_data(values.to_data())
.build()
.expect("compacted list array should contain valid data");
Arc::new(GenericListArray::<$offset_type>::from(data)) as ArrayRef
} else {
Arc::new(<$array_type>::new(
Arc::clone($field),
offsets,
values,
list.nulls().cloned(),
)) as ArrayRef
}
}};
}
// Macro for the i32/i64-offset list-view pair (ListView / LargeListView).
Expand Down
64 changes: 56 additions & 8 deletions datafusion/common/src/utils/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,8 @@ use crate::error::{
};
use crate::{Result, ScalarValue};
use arrow::array::{
Array, ArrayRef, FixedSizeListArray, LargeListArray, ListArray, OffsetSizeTrait,
cast::AsArray,
Array, ArrayData, ArrayRef, FixedSizeListArray, LargeListArray, ListArray,
OffsetSizeTrait, cast::AsArray,
};
use arrow::array::{
ArrowPrimitiveType, BooleanArray, Datum, GenericListArray, Int32Array, Int64Array,
Expand Down Expand Up @@ -591,9 +591,7 @@ impl SingleRowListArrayBuilder {

/// Build a single element [`ListArray`]
pub fn build_list_array(self) -> ListArray {
let (field, arr) = self.into_field_and_arr();
let offsets = OffsetBuffer::from_lengths([arr.len()]);
ListArray::new(field, offsets, arr, None)
self.build_generic_list_array()
}

/// Build a single element [`ListArray`] and wrap as [`ScalarValue::List`]
Expand All @@ -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);
Comment on lines +622 to +625

@sunchao sunchao Sep 9, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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.

@goutamadwant goutamadwant Sep 14, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@sunchao sunchao Sep 9, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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. Arrow's dictionary copy 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, 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.

@goutamadwant goutamadwant Sep 14, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@sunchao sunchao Sep 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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.

}

GenericListArray::new(field, offsets, arr, None)
}

/// Build a single element [`LargeListArray`] and wrap as [`ScalarValue::LargeList`]
Expand Down Expand Up @@ -1528,9 +1548,9 @@ mod tests {
use super::*;
use crate::ScalarValue::Null;
use arrow::{
array::{Float64Array, Int32Array},
array::{DictionaryArray, Float64Array, Int8Array, Int32Array, StringArray},
buffer::NullBuffer,
datatypes::Int32Type,
datatypes::{Int8Type, Int32Type},
};
#[cfg(feature = "sql")]
use sqlparser::ast::Ident;
Expand Down Expand Up @@ -1572,6 +1592,34 @@ mod tests {
}
}

#[test]
fn single_row_list_builder_handles_conservatively_nullable_values() {
let keys = Int8Array::from(vec![0, 0]);
let dictionary_values = Arc::new(StringArray::from(vec![Some("a"), None]));
let values = Arc::new(
DictionaryArray::<Int8Type>::try_new(keys, dictionary_values).unwrap(),
) as ArrayRef;

assert_eq!(values.logical_null_count(), 0);
assert!(values.is_nullable());

let list = SingleRowListArrayBuilder::new(Arc::clone(&values))
.with_nullable(false)
.build_list_array();
list.to_data().validate_full().unwrap();

let value = ScalarValue::try_from_array(&list, 0).unwrap();
assert_eq!(value.data_type(), list.data_type().clone());

let list = SingleRowListArrayBuilder::new(values)
.with_nullable(false)
.build_large_list_array();
list.to_data().validate_full().unwrap();

let value = ScalarValue::try_from_array(&list, 0).unwrap();
assert_eq!(value.data_type(), list.data_type().clone());
}

#[test]
fn test_bisect_linear_left_and_right() -> Result<()> {
let arrays: Vec<ArrayRef> = vec![
Expand Down
1 change: 1 addition & 0 deletions datafusion/spark/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ name = "datafusion_spark"

[dependencies]
arrow = { workspace = true }
arrow-select = { workspace = true }
base64 = "0.23"
bigdecimal = { workspace = true }
chrono = { workspace = true }
Expand Down
Loading
Loading