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/09/27 14:39:51 UTC
(datafusion-comet) 01/01: test: cover matching nested nulls in collect aggregates (#5970)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5970-605051ad239ef704f5f25d67910a446a6b6d7c70
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 7d2ea811edbd4eea31b1b31478a6c63b1b0278e8
Author: Chao Sun <su...@apache.org>
AuthorDate: Sun Sep 27 14:39:26 2026 +0000
test: cover matching nested nulls in collect aggregates (#5970)
---
native/core/src/execution/planner.rs | 69 ++++++++++++++++++++++
.../apache/comet/exec/CometAggregateSuite.scala | 26 ++++++++
2 files changed, 95 insertions(+)
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 80f8d890a3..91cc173818 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -7438,6 +7438,75 @@ mod tests {
assert!(Arc::ptr_eq(&raw, &coerced));
}
+ /// Normalizing an already-matching typed-null scalar or array must preserve its shared
+ /// storage, and both scalar collect accumulators must discard the null and return an empty
+ /// list. Duplicate field names with different types detect accidental name-based struct
+ /// conversion; matching runtime types must take DataFusion's identity cast path instead.
+ #[test]
+ fn test_collect_agg_preserves_matching_nested_nulls() {
+ use datafusion::logical_expr::ColumnarValue;
+ use datafusion::physical_expr::expressions::Literal;
+
+ let struct_type = DataType::Struct(Fields::from(vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new(
+ "a",
+ DataType::Struct(Fields::from(vec![Field::new("x", DataType::Int32, true)])),
+ true,
+ ),
+ ]));
+ let null_list = Arc::new(ListArray::new_null(
+ Arc::new(Field::new("item", struct_type, true)),
+ 1,
+ ));
+ let null_array = Arc::clone(&null_list) as ArrayRef;
+ let schema = collect_agg_schema(null_array.data_type().clone());
+ let batch =
+ RecordBatch::try_new(Arc::clone(&schema), vec![Arc::clone(&null_array)]).unwrap();
+
+ for raw in [
+ Arc::new(Literal::new(ScalarValue::List(Arc::clone(&null_list))))
+ as Arc<dyn PhysicalExpr>,
+ Arc::new(Column::new("s", 0)) as Arc<dyn PhysicalExpr>,
+ ] {
+ let child = PhysicalPlanner::coerce_collect_child_nullability(raw, &schema).unwrap();
+ let value = child.evaluate(&batch).unwrap();
+ match &value {
+ ColumnarValue::Scalar(ScalarValue::List(list)) => {
+ assert!(Arc::ptr_eq(list, &null_list));
+ }
+ ColumnarValue::Array(array) => assert!(Arc::ptr_eq(array, &null_array)),
+ other => panic!("expected a list scalar or array, got {other:?}"),
+ }
+ let arg = value.into_array(batch.num_rows()).unwrap();
+
+ for func in [
+ AggregateUDF::new_from_impl(CometCollectList::new()),
+ AggregateUDF::new_from_impl(CometCollectSet::new()),
+ ] {
+ let agg = PhysicalPlanner::create_aggr_func_expr(
+ "collect",
+ Arc::clone(&schema),
+ vec![Arc::clone(&child)],
+ func,
+ )
+ .unwrap();
+ let mut acc = agg.create_accumulator().unwrap();
+ acc.update_batch(&[Arc::clone(&arg)]).unwrap();
+ assert_eq!(
+ acc.state().unwrap()[0].data_type(),
+ *agg.state_fields().unwrap()[0].data_type()
+ );
+ let ScalarValue::List(result) = acc.evaluate().unwrap() else {
+ panic!("expected a list result");
+ };
+ assert_eq!(result.len(), 1);
+ assert!(!result.is_null(0));
+ assert!(result.value(0).is_empty());
+ }
+ }
+ }
+
#[test]
fn test_metadata_field_id_constants_match_iceberg_rust() {
// These constants are duplicated in Scala (CometIcebergNativeScan.MetadataFieldIds)
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
index b4ef6f3cc8..067b2d6437 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
@@ -211,6 +211,32 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
+ for (grouped <- Seq(false, true)) {
+ test(s"collect_list/collect_set over typed nested NULL (grouped=$grouped)") {
+ // A folded typed NULL reaches native collection normalization as a scalar. Both scalar
+ // and grouped accumulators must ignore the NULL and return empty collections with the
+ // declared nested element type. Require executed native Partial/Final aggregates so that
+ // constant folding or Spark fallback cannot hide the normalization path.
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+ withParquetTable(Seq((0, 1), (0, 2), (1, 3)), "tbl") {
+ val typedNull = "CAST(NULL AS ARRAY<STRUCT<a: INT, b: STRUCT<x: INT>>>)"
+ val projection = if (grouped) "_1, " else ""
+ val grouping = if (grouped) " GROUP BY _1" else ""
+ val (_, cometPlan) = checkSparkAnswerAndOperator(
+ sql(s"SELECT ${projection}collect_list($typedNull), collect_set($typedNull) " +
+ s"FROM tbl$grouping"))
+ val aggregates = collect(cometPlan) { case aggregate: CometHashAggregateExec =>
+ aggregate
+ }
+ assert(aggregates.exists(_.modes.contains(Partial)))
+ assert(aggregates.exists(_.modes.contains(Final)))
+ }
+ }
+ }
+ }
+
test("collect_list/collect_set combined with distinct aggregate falls back safely") {
// SPARK-17616: a distinct aggregate combined with collect_list/collect_set produces a
// multi-stage plan where the buffer-producing Partial may run in Spark (e.g. over a
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org