Skip to content

fix: ensure new_list respects data_type argument #24029

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
Ruchirtripathi wants to merge 8 commits into
base: main
Choose a base branch
from
+167 −16
Open

fix: ensure new_list respects data_type argument #24029

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
3c014b9
fix: ensure new_list respects data_type argument
Ruchirtripathi Jul 31, 2026
1239b46
fix: enforce strict cast in new_list and bypass sloppy test builder s…
Ruchirtripathi Aug 1, 2026
a02d17a
Merge branch 'apache:main' into fix-scalar-value-new-list
Ruchirtripathi Aug 1, 2026
7f0ef8a
Merge branch 'main' into fix-scalar-value-new-list
Ruchirtripathi Aug 1, 2026
7bd3221
fix(array_agg): correctly declare test accumulator schema
Ruchirtripathi Aug 4, 2026
e0b18a5
style: run cargo fmt
Ruchirtripathi Aug 4, 2026
2bdbd92
fix(test): restore missing test annotation for desc_order_partial_fin…
Ruchirtripathi Aug 4, 2026
551ea00
test: add nested nullability regression test for DISTINCT ARRAY_AGG
Ruchirtripathi Aug 4, 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
33 changes: 30 additions & 3 deletions datafusion/common/src/scalar/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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())
}
Expand Down Expand Up @@ -11544,4 +11547,28 @@ mod tests {
run_tests::<Decimal128Type>();
run_tests::<Decimal256Type>();
}

#[test]
fn test_new_list_nested_nullability_mismatch_issue_24022() {
Comment thread
Ruchirtripathi marked this conversation as resolved.
// 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);

@kosiew kosiew Aug 4, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Could we extend this regression to cover all three constructors changed by this PR?

Right now it only calls new_list and checks list.data_type(). A shared assertion for new_list, new_list_from_iter, and new_large_list would also let us verify both the declared nested child type and the normalized child values.

This is a suggestion only.

assert_eq!(
list.data_type(),
&DataType::List(Arc::new(Field::new_list_field(
requested_element_type,
true
)))
);
}
}
150 changes: 137 additions & 13 deletions datafusion/functions-aggregate/src/array_agg.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1741,15 +1741,17 @@ 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()
.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"],
Expand All @@ -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"],
Expand All @@ -1781,6 +1785,122 @@ mod tests {
Ok(())
}

#[test]
fn ordered_aggregate_nested_nullability_mismatch_issue_24022() -> Result<()> {

@kosiew kosiew Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Thanks for adding the ordered aggregate regression. Could you also add equivalent coverage for DistinctArrayAggAccumulator::evaluate?

Please use a declared struct type with a nullable field and runtime values whose struct field is non-nullable, then assert both the exact resulting list type and the values. This is the remaining blocking item because the DISTINCT decoding path can preserve different nested nullability.

@Ruchirtripathi Ruchirtripathi Aug 4, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Thanks for the detailed review! That makes sense. I'll add an equivalent regression for DistinctArrayAggAccumulator::evaluate using a declared struct type with a nullable field and runtime values whose struct field is non-nullable. I'll assert both the exact resulting list type and the values to cover the nested nullability preservation in the DISTINCT decoding path, then update the PR.

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
Expand Down Expand Up @@ -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(),
},
}
Expand Down
Loading