From 3c014b92d7b84ca7c90a78f1611f65ceb82d2d52 Mon Sep 17 00:00:00 2001 From: Ruchirtripathi Date: Fri, 31 Jul 2026 19:49:41 +0530 Subject: [PATCH 1/6] fix: ensure new_list respects data_type argument --- datafusion/common/src/scalar/mod.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs index 924620a930869..400ab441afaa3 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()) } From 1239b46ed03a09343f5681102cea2b191ec74c78 Mon Sep 17 00:00:00 2001 From: Ruchirtripathi Date: Sat, 1 Aug 2026 07:57:01 +0530 Subject: [PATCH 2/6] fix: enforce strict cast in new_list and bypass sloppy test builder schema in array_agg accumulators --- datafusion/common/src/scalar/mod.rs | 24 +++++++++++++++++++ .../functions-aggregate/src/array_agg.rs | 7 +++--- 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs index 400ab441afaa3..60da66383a556 100644 --- a/datafusion/common/src/scalar/mod.rs +++ b/datafusion/common/src/scalar/mod.rs @@ -11547,4 +11547,28 @@ mod tests { run_tests::(); run_tests::(); } + + #[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 list = ScalarValue::new_list(&[value], &requested_element_type, true); + assert_eq!( + list.data_type(), + &DataType::List(Arc::new(Field::new_list_field( + requested_element_type, + true + ))) + ); + } } diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs index cfacd771968c2..2445ed5cf4a8e 100644 --- a/datafusion/functions-aggregate/src/array_agg.rs +++ b/datafusion/functions-aggregate/src/array_agg.rs @@ -1059,7 +1059,7 @@ impl Accumulator for DistinctArrayAggAccumulator { .map(|i| ScalarValue::try_from_array(decoded.as_ref(), i)) .collect::>()?; - let arr = ScalarValue::new_list(&values, &self.datatype, true); + let arr = ScalarValue::new_list(&values, decoded.data_type(), true); Ok(ScalarValue::List(arr)) } @@ -1429,14 +1429,15 @@ impl Accumulator for OrderSensitiveArrayAggAccumulator { } let values = self.values.clone(); + let values_data_type = values[0].data_type(); let array = if self.reverse { ScalarValue::new_list_from_iter( values.into_iter().rev(), - &self.datatypes[0], + &values_data_type, true, ) } else { - ScalarValue::new_list_from_iter(values.into_iter(), &self.datatypes[0], true) + ScalarValue::new_list_from_iter(values.into_iter(), &values_data_type, true) }; Ok(ScalarValue::List(array)) } From 7bd32211764cf50f04d0079399e8e76a765729b9 Mon Sep 17 00:00:00 2001 From: Ruchirtripathi Date: Tue, 4 Aug 2026 17:55:00 +0530 Subject: [PATCH 3/6] fix(array_agg): correctly declare test accumulator schema --- .../functions-aggregate/src/array_agg.rs | 86 +++++++++++++++---- 1 file changed, 68 insertions(+), 18 deletions(-) diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs index 2445ed5cf4a8e..286f2ad4c71b0 100644 --- a/datafusion/functions-aggregate/src/array_agg.rs +++ b/datafusion/functions-aggregate/src/array_agg.rs @@ -1059,7 +1059,7 @@ impl Accumulator for DistinctArrayAggAccumulator { .map(|i| ScalarValue::try_from_array(decoded.as_ref(), i)) .collect::>()?; - let arr = ScalarValue::new_list(&values, decoded.data_type(), true); + let arr = ScalarValue::new_list(&values, &self.datatype, true); Ok(ScalarValue::List(arr)) } @@ -1429,15 +1429,14 @@ impl Accumulator for OrderSensitiveArrayAggAccumulator { } let values = self.values.clone(); - let values_data_type = values[0].data_type(); let array = if self.reverse { ScalarValue::new_list_from_iter( values.into_iter().rev(), - &values_data_type, + &self.datatypes[0], true, ) } else { - ScalarValue::new_list_from_iter(values.into_iter(), &values_data_type, true) + ScalarValue::new_list_from_iter(values.into_iter(), &self.datatypes[0], true) }; Ok(ScalarValue::List(array)) } @@ -1742,13 +1741,13 @@ mod tests { acc2.update_batch(&[data(["b", "c", "a"])])?; acc1 = merge(acc1, acc2)?; - assert_eq!(acc1.size(), 282); + assert_eq!(acc1.size(), 166); Ok(()) } #[test] fn does_not_over_account_memory_distinct() -> Result<()> { - let (mut acc1, mut acc2) = ArrayAggAccumulatorBuilder::string() + let (mut acc1, mut acc2) = ArrayAggAccumulatorBuilder::new(DataType::List(Arc::new(Field::new_list_field(DataType::Utf8, true)))) .distinct() .build_two()?; @@ -1766,7 +1765,7 @@ mod tests { #[test] fn does_not_over_account_memory_ordered() -> Result<()> { - let mut acc = ArrayAggAccumulatorBuilder::string() + 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()?; @@ -1782,12 +1781,62 @@ mod tests { 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 - // a final accumulator via merge_batch; without the fix the ordering keys and - // values are mismatched so the final sort produces wrong order. #[test] + fn ordered_aggregate_nested_nullability_mismatch_issue_24022() -> Result<()> { + use datafusion_physical_expr::expressions::Column; + use arrow::array::{StructArray, Int32Array, Int64Array}; + + 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; + + 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()?; + + assert_eq!( + evaluated.data_type(), + DataType::List(Arc::new(Field::new_list_field(requested_element_type, true))) + ); + + Ok(()) + } + + // Reproduces the bug where `state()` emits reversed values but non-reversed fn desc_order_partial_final_merge_correct() -> Result<()> { use arrow::array::Int64Array; use datafusion_physical_expr::expressions::Column; @@ -1905,15 +1954,16 @@ 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(), }, } From e0b18a5108112e138c383b3d1facacfb648685c5 Mon Sep 17 00:00:00 2001 From: Ruchirtripathi Date: Tue, 4 Aug 2026 17:58:23 +0530 Subject: [PATCH 4/6] style: run cargo fmt --- .../functions-aggregate/src/array_agg.rs | 39 ++++++++++++------- 1 file changed, 25 insertions(+), 14 deletions(-) diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs index 286f2ad4c71b0..8a9d2120c4d4e 100644 --- a/datafusion/functions-aggregate/src/array_agg.rs +++ b/datafusion/functions-aggregate/src/array_agg.rs @@ -1747,9 +1747,11 @@ mod tests { } #[test] fn does_not_over_account_memory_distinct() -> Result<()> { - let (mut acc1, mut acc2) = ArrayAggAccumulatorBuilder::new(DataType::List(Arc::new(Field::new_list_field(DataType::Utf8, true)))) - .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::new(DataType::List(Arc::new(Field::new_list_field(DataType::Utf8, true)))) - .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"], @@ -1783,9 +1787,9 @@ mod tests { #[test] fn ordered_aggregate_nested_nullability_mismatch_issue_24022() -> Result<()> { + use arrow::array::{Int32Array, Int64Array, StructArray}; use datafusion_physical_expr::expressions::Column; - use arrow::array::{StructArray, Int32Array, Int64Array}; - + let requested_element_type = DataType::Struct(Fields::from(vec![Field::new("n", DataType::Int32, true)])); let inferred_field = Field::new("n", DataType::Int32, false); @@ -1798,7 +1802,7 @@ mod tests { let ord_expr = Arc::new( Column::new_with_schema("ord", &schema).expect("column not in schema"), ) as Arc; - + let asc_opts = SortOptions { descending: false, nulls_first: false, @@ -1806,7 +1810,8 @@ mod tests { let asc_ordering = LexOrdering::new(vec![PhysicalSortExpr::new( Arc::clone(&ord_expr), asc_opts, - )]).unwrap(); + )]) + .unwrap(); let mut acc = OrderSensitiveArrayAggAccumulator::try_new( &requested_element_type, @@ -1825,12 +1830,15 @@ mod tests { let ord_arr = Arc::new(Int64Array::from(vec![0i64])) as ArrayRef; acc.update_batch(&[value_arr, ord_arr])?; - + let evaluated = acc.evaluate()?; - + assert_eq!( evaluated.data_type(), - DataType::List(Arc::new(Field::new_list_field(requested_element_type, true))) + DataType::List(Arc::new(Field::new_list_field( + requested_element_type, + true + ))) ); Ok(()) @@ -1956,7 +1964,10 @@ mod tests { Self { return_field: Field::new( "f", - DataType::List(Arc::new(Field::new_list_field(data_type.clone(), true))), + DataType::List(Arc::new(Field::new_list_field( + data_type.clone(), + true, + ))), true, ) .into(), From 2bdbd928ad309e4073a990520a37ef742960532f Mon Sep 17 00:00:00 2001 From: Ruchirtripathi Date: Tue, 4 Aug 2026 18:16:26 +0530 Subject: [PATCH 5/6] fix(test): restore missing test annotation for desc_order_partial_final_merge_correct --- datafusion/functions-aggregate/src/array_agg.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs index 8a9d2120c4d4e..b7504c3409629 100644 --- a/datafusion/functions-aggregate/src/array_agg.rs +++ b/datafusion/functions-aggregate/src/array_agg.rs @@ -1845,6 +1845,11 @@ mod tests { } // 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 + // a final accumulator via merge_batch; without the fix the ordering keys and + // values are mismatched so the final sort produces wrong order. + #[test] fn desc_order_partial_final_merge_correct() -> Result<()> { use arrow::array::Int64Array; use datafusion_physical_expr::expressions::Column; From 551ea00645fe3c2daae277e8e20e82b8e8fdd755 Mon Sep 17 00:00:00 2001 From: Ruchirtripathi Date: Tue, 4 Aug 2026 20:53:33 +0530 Subject: [PATCH 6/6] test: add nested nullability regression test for DISTINCT ARRAY_AGG --- .../functions-aggregate/src/array_agg.rs | 71 +++++++++++++++++-- 1 file changed, 64 insertions(+), 7 deletions(-) diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs index b7504c3409629..e15ddcd825d2f 100644 --- a/datafusion/functions-aggregate/src/array_agg.rs +++ b/datafusion/functions-aggregate/src/array_agg.rs @@ -1833,13 +1833,70 @@ mod tests { let evaluated = acc.evaluate()?; - assert_eq!( - evaluated.data_type(), - DataType::List(Arc::new(Field::new_list_field( - requested_element_type, - true - ))) - ); + 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(()) }