You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/12 10:31:39 UTC
(datafusion) branch main updated: fix: ensure new_list respects data_type argument (#24029)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 9b3b518950 fix: ensure new_list respects data_type argument (#24029)
9b3b518950 is described below
commit 9b3b518950c0c6a564b5525d6b274000f27dccd9
Author: Ruchir Tripathi <16...@users.noreply.github.com>
AuthorDate: Wed Aug 12 10:19:11 2026 +0000
fix: ensure new_list respects data_type argument (#24029)
## Which issue does this PR close?
- Closes #24022
## Rationale for this change
Fixes a bug where `ScalarValue::new_list`, `new_list_nullable`, and
`new_large_list` silently ignored the
`data_type` argument when the `values` array was non-empty. This caused
issues where accumulators like
`collect_list` could produce outputs with a slightly different type than
declared (e.g., in the nullability of
nested fields), leading to invalid argument errors in
`GroupedHashAggregateStream::emit`.
## What changes are included in this PR?
- Added a `cast_with_options` call for non-empty lists in
`ScalarValue::new_list`, `new_list_from_iter`, and
`new_large_list`.
- Used `DEFAULT_CAST_OPTIONS` to ensure the concatenated array is
properly reconciled with the requested
`data_type`.
## Are these changes tested?
Yes, this is covered by existing tests. It resolves the
`GroupedHashAggregateStream` output batch validation
failures for accumulators.
## Are there any user-facing changes?
No, this is an internal bug fix.
---
datafusion/common/src/scalar/mod.rs | 72 +++++++++++-
datafusion/functions-aggregate/src/array_agg.rs | 150 ++++++++++++++++++++++--
2 files changed, 206 insertions(+), 16 deletions(-)
diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs
index 924620a930..cb0442392a 100644
--- a/datafusion/common/src/scalar/mod.rs
+++ b/datafusion/common/src/scalar/mod.rs
@@ -3309,7 +3309,8 @@ impl ScalarValue {
let values = if values.is_empty() {
new_empty_array(data_type)
} else {
- Self::iter_to_array(values.iter().cloned()).unwrap()
+ let arr = Self::iter_to_array(values.iter().cloned()).unwrap();
+ cast_with_options(&arr, data_type, &DEFAULT_CAST_OPTIONS).unwrap()
};
Arc::new(
SingleRowListArrayBuilder::new(values)
@@ -3371,7 +3372,8 @@ impl ScalarValue {
let values = if values.len() == 0 {
new_empty_array(data_type)
} else {
- Self::iter_to_array(values).unwrap()
+ let arr = Self::iter_to_array(values).unwrap();
+ cast_with_options(&arr, data_type, &DEFAULT_CAST_OPTIONS).unwrap()
};
Arc::new(
SingleRowListArrayBuilder::new(values)
@@ -3414,7 +3416,8 @@ impl ScalarValue {
let values = if values.is_empty() {
new_empty_array(data_type)
} else {
- Self::iter_to_array(values.iter().cloned()).unwrap()
+ let arr = Self::iter_to_array(values.iter().cloned()).unwrap();
+ cast_with_options(&arr, data_type, &DEFAULT_CAST_OPTIONS).unwrap()
};
Arc::new(SingleRowListArrayBuilder::new(values).build_large_list_array())
}
@@ -11544,4 +11547,67 @@ mod tests {
run_tests::<Decimal128Type>();
run_tests::<Decimal256Type>();
}
+
+ #[test]
+ fn test_new_list_nested_nullability_mismatch_issue_24022() {
+ // requested element type: Struct(n: Int32 nullable=true)
+ let requested_element_type =
+ DataType::Struct(Fields::from(vec![Field::new("n", DataType::Int32, true)]));
+
+ // inferred from concrete values: Struct(n: Int32 nullable=false)
+ let inferred_field = Field::new("n", DataType::Int32, false);
+
+ let value = ScalarValue::Struct(Arc::new(StructArray::from(vec![(
+ Arc::new(inferred_field),
+ Arc::new(Int32Array::from(vec![1])) as ArrayRef,
+ )])));
+
+ let expected_struct_array = StructArray::from(vec![(
+ Arc::new(Field::new("n", DataType::Int32, true)),
+ Arc::new(Int32Array::from(vec![1])) as ArrayRef,
+ )]);
+ let expected_array = Arc::new(expected_struct_array) as ArrayRef;
+
+ // Test new_list
+ let list = ScalarValue::new_list(
+ std::slice::from_ref(&value),
+ &requested_element_type,
+ true,
+ );
+ assert_eq!(
+ list.data_type(),
+ &DataType::List(Arc::new(Field::new_list_field(
+ requested_element_type.clone(),
+ true
+ )))
+ );
+ assert_eq!(&list.value(0), &expected_array);
+
+ // Test new_list_from_iter
+ let list_from_iter = ScalarValue::new_list_from_iter(
+ std::iter::once(value.clone()),
+ &requested_element_type,
+ true,
+ );
+ assert_eq!(
+ list_from_iter.data_type(),
+ &DataType::List(Arc::new(Field::new_list_field(
+ requested_element_type.clone(),
+ true
+ )))
+ );
+ assert_eq!(&list_from_iter.value(0), &expected_array);
+
+ // Test new_large_list
+ let large_list = ScalarValue::new_large_list(&[value], &requested_element_type);
+ assert_eq!(
+ large_list.data_type(),
+ &DataType::LargeList(Arc::new(Field::new(
+ "item",
+ requested_element_type.clone(),
+ true
+ )))
+ );
+ assert_eq!(&large_list.value(0), &expected_array);
+ }
}
diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs
index eaf7f9addb..0e02ff1186 100644
--- a/datafusion/functions-aggregate/src/array_agg.rs
+++ b/datafusion/functions-aggregate/src/array_agg.rs
@@ -1741,15 +1741,17 @@ mod tests {
acc2.update_batch(&[data(["b", "c", "a"])])?;
acc1 = merge(acc1, acc2)?;
- assert_eq!(acc1.size(), 290);
+ assert_eq!(acc1.size(), 174);
Ok(())
}
#[test]
fn does_not_over_account_memory_distinct() -> Result<()> {
- let (mut acc1, mut acc2) = ArrayAggAccumulatorBuilder::string()
- .distinct()
- .build_two()?;
+ let (mut acc1, mut acc2) = ArrayAggAccumulatorBuilder::new(DataType::List(
+ Arc::new(Field::new_list_field(DataType::Utf8, true)),
+ ))
+ .distinct()
+ .build_two()?;
acc1.update_batch(&[string_list_data([
vec!["a", "b", "c"],
@@ -1765,9 +1767,11 @@ mod tests {
#[test]
fn does_not_over_account_memory_ordered() -> Result<()> {
- let mut acc = ArrayAggAccumulatorBuilder::string()
- .order_by_col("col", SortOptions::new(false, false))
- .build()?;
+ let mut acc = ArrayAggAccumulatorBuilder::new(DataType::List(Arc::new(
+ Field::new_list_field(DataType::Utf8, true),
+ )))
+ .order_by_col("col", SortOptions::new(false, false))
+ .build()?;
acc.update_batch(&[string_list_data([
vec!["a", "b", "c"],
@@ -1781,6 +1785,122 @@ mod tests {
Ok(())
}
+ #[test]
+ fn ordered_aggregate_nested_nullability_mismatch_issue_24022() -> Result<()> {
+ use arrow::array::{Int32Array, Int64Array, StructArray};
+ use datafusion_physical_expr::expressions::Column;
+
+ let requested_element_type =
+ DataType::Struct(Fields::from(vec![Field::new("n", DataType::Int32, true)]));
+ let inferred_field = Field::new("n", DataType::Int32, false);
+
+ let ordering_dtype = DataType::Int64;
+ let schema = Schema::new(vec![
+ Field::new("val", requested_element_type.clone(), true),
+ Field::new("ord", DataType::Int64, true),
+ ]);
+ let ord_expr = Arc::new(
+ Column::new_with_schema("ord", &schema).expect("column not in schema"),
+ ) as Arc<dyn PhysicalExpr>;
+
+ let asc_opts = SortOptions {
+ descending: false,
+ nulls_first: false,
+ };
+ let asc_ordering = LexOrdering::new(vec![PhysicalSortExpr::new(
+ Arc::clone(&ord_expr),
+ asc_opts,
+ )])
+ .unwrap();
+
+ let mut acc = OrderSensitiveArrayAggAccumulator::try_new(
+ &requested_element_type,
+ std::slice::from_ref(&ordering_dtype),
+ asc_ordering,
+ /*is_input_pre_ordered=*/ true,
+ /*reverse=*/ false,
+ /*ignore_nulls=*/ false,
+ )?;
+
+ let value_arr = Arc::new(StructArray::from(vec![(
+ Arc::new(inferred_field),
+ Arc::new(Int32Array::from(vec![1])) as ArrayRef,
+ )])) as ArrayRef;
+
+ let ord_arr = Arc::new(Int64Array::from(vec![0i64])) as ArrayRef;
+
+ acc.update_batch(&[value_arr, ord_arr])?;
+
+ let evaluated = acc.evaluate()?;
+
+ if let ScalarValue::List(arr) = evaluated {
+ assert_eq!(
+ arr.data_type(),
+ &DataType::List(Arc::new(Field::new_list_field(
+ requested_element_type.clone(),
+ true
+ )))
+ );
+
+ let expected_struct_array = StructArray::from(vec![(
+ Arc::new(Field::new("n", DataType::Int32, true)),
+ Arc::new(Int32Array::from(vec![1])) as ArrayRef,
+ )]);
+ let expected_array = Arc::new(expected_struct_array) as ArrayRef;
+ assert_eq!(&arr.value(0), &expected_array);
+ } else {
+ panic!("Expected ScalarValue::List");
+ }
+
+ Ok(())
+ }
+
+ #[test]
+ fn distinct_aggregate_nested_nullability_mismatch_issue_24022() -> Result<()> {
+ use arrow::array::{Int32Array, StructArray};
+ use datafusion_common::ScalarValue;
+
+ let requested_element_type =
+ DataType::Struct(Fields::from(vec![Field::new("n", DataType::Int32, true)]));
+ let inferred_field = Field::new("n", DataType::Int32, false);
+
+ let mut acc = DistinctArrayAggAccumulator::try_new(
+ &requested_element_type,
+ None,
+ /*ignore_nulls=*/ false,
+ )?;
+
+ let value_arr = Arc::new(StructArray::from(vec![(
+ Arc::new(inferred_field),
+ Arc::new(Int32Array::from(vec![1])) as ArrayRef,
+ )])) as ArrayRef;
+
+ acc.update_batch(&[value_arr])?;
+
+ let evaluated = acc.evaluate()?;
+
+ if let ScalarValue::List(arr) = evaluated {
+ assert_eq!(
+ arr.data_type(),
+ &DataType::List(Arc::new(Field::new_list_field(
+ requested_element_type.clone(),
+ true
+ )))
+ );
+
+ let expected_struct_array = StructArray::from(vec![(
+ Arc::new(Field::new("n", DataType::Int32, true)),
+ Arc::new(Int32Array::from(vec![1])) as ArrayRef,
+ )]);
+ let expected_array = Arc::new(expected_struct_array) as ArrayRef;
+ assert_eq!(&arr.value(0), &expected_array);
+ } else {
+ panic!("Expected ScalarValue::List");
+ }
+
+ Ok(())
+ }
+
// Reproduces the bug where `state()` emits reversed values but non-reversed
// orderings when the optimizer sets is_input_pre_ordered=true + reverse=true
// (DESC aggregate with ASC pre-sorted input). The partial states are fed into
@@ -1904,15 +2024,19 @@ mod tests {
fn new(data_type: DataType) -> Self {
Self {
- return_field: Field::new("f", data_type.clone(), true).into(),
+ return_field: Field::new(
+ "f",
+ DataType::List(Arc::new(Field::new_list_field(
+ data_type.clone(),
+ true,
+ ))),
+ true,
+ )
+ .into(),
distinct: false,
order_bys: vec![],
schema: Schema {
- fields: Fields::from(vec![Field::new(
- "col",
- DataType::new_list(data_type, true),
- true,
- )]),
+ fields: Fields::from(vec![Field::new("col", data_type, true)]),
metadata: Default::default(),
},
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org