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/23 13:01:35 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-4816-cccc08b7cee91fc4a3ec4ea4236dbc57906ccdd5 created (now ef7bda56be)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-4816-cccc08b7cee91fc4a3ec4ea4236dbc57906ccdd5
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at ef7bda56be feat: support listagg / string_agg aggregate (Spark 4.0+) (#4816)
This branch includes the following new commits:
new ef7bda56be feat: support listagg / string_agg aggregate (Spark 4.0+) (#4816)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: feat: support listagg / string_agg aggregate (Spark 4.0+) (#4816)
Posted by gi...@apache.org.
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-4816-cccc08b7cee91fc4a3ec4ea4236dbc57906ccdd5
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit ef7bda56be6114738bb316b7896c5dc05267e5a3
Author: Andy Grove <ag...@apache.org>
AuthorDate: Wed Sep 23 13:01:10 2026 +0000
feat: support listagg / string_agg aggregate (Spark 4.0+) (#4816)
* feat: support listagg / string_agg aggregate (Spark 4.0+)
Adds a `SparkListAgg` UDAF and a `CometListAgg` serde so that Comet can
natively execute the simple form of Spark 4.0's `LISTAGG(child, delimiter)`
/ `string_agg` on `StringType` inputs with a literal delimiter. `WITHIN
GROUP (ORDER BY ...)`, `BinaryType` inputs, non-literal delimiters, and
non-default collations fall back to Spark. `DISTINCT` falls back because
Comet already rejects multi-column distinct aggregates.
The native accumulator returns `Utf8` but keeps its intermediate state as
`Binary` to match Spark's `TypedImperativeAggregate` buffer schema so the
Comet shuffle layer does not insert a `Utf8` β `Binary` cast the merge
side cannot read back.
Scaffolding produced by the `implement-comet-expression` skill.
* refactor: hoist CometListAgg reason strings and use `.foldable`
Consolidate the reason strings into `private val`s so `getSupportLevel`
and `getUnsupportedReasons` reference the same source of truth, and
replace the manual `case _: Literal` delimiter check with the standard
`.foldable` gate used elsewhere (e.g. `CometPercentile`).
* refactor: address listagg review feedback
Add a GroupsAccumulator fast path to SparkListAgg for grouped aggregation,
remove the dead `datatype` proto field and unreachable serialization branch,
drop the never-emitted `distinctReason`, and add a multi-partition merge test.
Keep the `Binary` intermediate state: Spark's `ListAgg` is a
`TypedImperativeAggregate` whose `aggBufferAttributes` is `BinaryType`, and the
shuffle Exchange between the partial and final aggregate uses that Spark-declared
schema even though partial and final run in the same engine. The module doc and
`state_fields` comment are rewritten to explain this.
* test: cover listagg collation fallback and align docs with the Binary state
- Add a UTF8_LCASE collation fallback case to listagg.sql; the aggregate
serde's collation guard is reachable even though the scan falls back.
- Drop the comments claiming the collation guard is unreachable.
- Describe DISTINCT as falling back via the multi-column distinct check in
aggExprToProto, not Spark's multi-stage rewrite (serde, native, proto).
- Update the audit doc: intermediate state is Binary to match Spark's
TypedImperativeAggregate buffer at the Exchange.
---
.../expression-audits/agg_funcs.md | 8 +
docs/source/user-guide/latest/expressions.md | 4 +-
native/core/src/execution/planner.rs | 10 +-
native/proto/src/proto/expr.proto | 15 +
native/spark-expr/src/agg_funcs/list_agg.rs | 480 +++++++++++++++++++++
native/spark-expr/src/agg_funcs/mod.rs | 2 +
.../org/apache/comet/serde/QueryPlanSerde.scala | 67 +--
.../org/apache/comet/shims/CometExprShim.scala | 4 +-
.../org/apache/comet/shims/CometExprShim.scala | 4 +-
.../org/apache/comet/serde/CometListAgg.scala | 99 +++++
.../apache/comet/shims/Spark4xCometExprShim.scala | 5 +-
.../sql-tests/expressions/aggregate/listagg.sql | 192 +++++++++
12 files changed, 852 insertions(+), 38 deletions(-)
diff --git a/docs/source/contributor-guide/expression-audits/agg_funcs.md b/docs/source/contributor-guide/expression-audits/agg_funcs.md
index 8d0ff7be68..75b7fbf225 100644
--- a/docs/source/contributor-guide/expression-audits/agg_funcs.md
+++ b/docs/source/contributor-guide/expression-audits/agg_funcs.md
@@ -72,6 +72,14 @@
- Performance (tuned 2026-09-09, PR [#5803](https://github.com/apache/datafusion-comet/pull/5803)): as for [collect_list](#collect-list), grouped `collect_set` no longer goes through `GroupsAccumulatorAdapter`. `CollectSetGroupsAccumulator` encodes each batch once for all of its groups and deduplicates against an open-addressed index over a shared arena, replacing one hash table plus one owned `Row` per distinct value per group. 60-93% faster ([#5797](https://github.com/apache/datafusion [...]
- `CometCollectSet` reports `Incompatible` for float and double input when `spark.comet.exec.strictFloatingPoint=true`, because the native distinct comparison treats `NaN == NaN` and collapses repeated `NaN`s into a single element while Spark keeps each one. The native path for floating-point input is then opt-in via `spark.comet.expression.CollectSet.allowIncompatible=true`. All other input types are `Compatible`.
+## listagg
+
+- Spark 3.4.3 (audited 2026-07-03): does not exist. `ListAgg` was added in Spark 4.0.
+- Spark 3.5.8 (audited 2026-07-03): does not exist.
+- Spark 4.0.1 (audited 2026-07-03): `ListAgg(child, delimiter, orderExpressions)` in `aggregate/collect.scala`. Accepts `StringType` or `BinaryType` inputs; result type matches child. Skips nulls; empty or all-null groups return `NULL`. A `NULL` delimiter is treated as an empty string. `CometListAgg` maps only the simple form: `StringType` child with a literal `StringType`/`NullType` delimiter and no `WITHIN GROUP`. `BinaryType` inputs, `WITHIN GROUP (ORDER BY ...)`, non-literal delimite [...]
+- Spark 4.1.1 (audited 2026-07-03): byte-identical to 4.0.1.
+- Native accumulator (`SparkListAgg`) returns `Utf8` and carries its intermediate state as `Binary`. Spark's `ListAgg` is a `TypedImperativeAggregate` with a single `BinaryType` buffer attribute, and the shuffle Exchange between the partial and final aggregate uses that schema, so the native state must be `Binary` even though partial and final always run in Comet (`supportsNativePartialToSparkFinal` and `supportsSparkPartialToNativeFinal` are false). A `GroupsAccumulator` fast path is pr [...]
+
## max_by
- Spark 3.4.3 (2026-07-03): `MaxBy` is a 2-argument `DeclarativeAggregate` registered as `expression[MaxBy]("max_by")`. Buffer is `(valueWithExtremumOrdering, extremumOrdering)`; null orderings are ignored, the value paired with the maximum ordering is returned (and may itself be null), and an all-null-ordering group yields null. Comet implements a native `max_by` aggregate. Only fixed-length value and ordering types are accelerated: a variable-length or nested type (string, binary, stru [...]
diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
index ab2794071a..700d027d0d 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/docs/source/user-guide/latest/expressions.md
@@ -107,7 +107,7 @@ The tables below list every Spark built-in expression with its current status.
| `kurtosis` | π | β | Not yet implemented natively |
| `last` | β
| Native | |
| `last_value` | β
| Native | |
-| `listagg` | π | β | String aggregation |
+| `listagg` | β
| Native | Spark 4.0+. `StringType` input with a literal delimiter; `WITHIN GROUP (ORDER BY ...)` and `BinaryType` inputs fall back to Spark. Without `WITHIN GROUP`, concatenation order is the (non-deterministic) group arrival order, matching Spark. |
| `max` | β
| Native | |
| `max_by` | β
| Native | Value and ordering must be fixed-length types |
| `mean` | β
| Native | |
@@ -133,7 +133,7 @@ The tables below list every Spark built-in expression with its current status.
| `stddev` | β
| Native | |
| `stddev_pop` | β
| Native | |
| `stddev_samp` | β
| Native | |
-| `string_agg` | π | β | String aggregation (alias of `listagg`) |
+| `string_agg` | β
| Native | Alias of `listagg`; same restrictions apply. |
| `sum` | β
| Native | |
| `try_avg` | β
| β | Interval types fall back |
| `try_sum` | β
| β | |
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 8f030da455..23c7c4bd02 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -93,7 +93,8 @@ use datafusion::{
use datafusion_comet_spark_expr::{
create_comet_physical_fun, create_comet_physical_fun_with_eval_mode, BinaryOutputStyle,
BloomFilterAgg, BloomFilterMightContain, CometCollectList, CometCollectSet, CsvWriteOptions,
- EvalMode, SparkArraysZipFunc, SparkBloomFilterVersion, SparkPercentile, SumInteger, ToCsv,
+ EvalMode, SparkArraysZipFunc, SparkBloomFilterVersion, SparkListAgg, SparkPercentile,
+ SumInteger, ToCsv,
};
use iceberg::expr::Bind;
@@ -3181,6 +3182,13 @@ impl PhysicalPlanner {
let func = AggregateUDF::new_from_impl(HllPlusPlus::new(expr.precision));
Self::create_aggr_func_expr("approx_count_distinct", schema, vec![child], func)
}
+ AggExprStruct::ListAgg(expr) => {
+ let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
+ let delimiter =
+ self.create_expr(expr.delimiter.as_ref().unwrap(), Arc::clone(&schema))?;
+ let func = AggregateUDF::new_from_impl(SparkListAgg::new());
+ Self::create_aggr_func_expr("listagg", schema, vec![child, delimiter], func)
+ }
AggExprStruct::MaxBy(expr) => {
let value = self.create_expr(expr.value.as_ref().unwrap(), Arc::clone(&schema))?;
let ordering =
diff --git a/native/proto/src/proto/expr.proto b/native/proto/src/proto/expr.proto
index 34d502daac..d96cf03de9 100644
--- a/native/proto/src/proto/expr.proto
+++ b/native/proto/src/proto/expr.proto
@@ -166,6 +166,7 @@ message AggExpr {
Mode mode = 23;
MaxBy maxBy = 24;
MinBy minBy = 25;
+ ListAgg listAgg = 26;
}
// Optional filter expression for SQL FILTER (WHERE ...) clause.
@@ -355,6 +356,20 @@ message HllPlusPlus {
int32 precision = 2;
}
+// Spark 4.0+ LISTAGG / STRING_AGG aggregate.
+//
+// Comet only serializes the simple form: a StringType child with a literal
+// (or NULL) delimiter and no WITHIN GROUP ORDER BY. DISTINCT falls back to
+// Spark because Comet rejects multi-column distinct aggregates in
+// aggExprToProto, so the native side never sees it.
+message ListAgg {
+ Expr child = 1;
+ // Literal delimiter expression. NULL delimiter is normalized to empty string
+ // by Spark's semantics. The result type is always Utf8 (Spark StringType), so
+ // no datatype field is needed.
+ Expr delimiter = 2;
+}
+
message MaxBy {
// The value returned by the aggregate (associated with the maximum ordering).
Expr value = 1;
diff --git a/native/spark-expr/src/agg_funcs/list_agg.rs b/native/spark-expr/src/agg_funcs/list_agg.rs
new file mode 100644
index 0000000000..eeda8c88ff
--- /dev/null
+++ b/native/spark-expr/src/agg_funcs/list_agg.rs
@@ -0,0 +1,480 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Spark-compatible `listagg` / `string_agg` aggregate function.
+//!
+//! Implements the simple form of Spark 4.0's `LISTAGG(expr, delimiter)` (no
+//! `WITHIN GROUP (ORDER BY ...)`, no DISTINCT; DISTINCT falls back to Spark
+//! because Comet rejects multi-column distinct aggregates in `aggExprToProto`).
+//! Differences from DataFusion's `string_agg`:
+//!
+//! * Returns `Utf8` to match Spark's `StringType` result type; DataFusion's
+//! `string_agg` returns `LargeUtf8`. Returning `Utf8` (not `LargeUtf8`) at
+//! the JVM FFI boundary is why this is a Comet-owned UDAF rather than a direct
+//! reuse of `string_agg`.
+//! * A `NULL` delimiter is treated as the empty string (Spark treats `NULL` as
+//! the default empty delimiter; the JVM serde forwards the literal as-is).
+//! * The delimiter is read once from the accumulator args (a literal is
+//! enforced by Spark's analyzer).
+//!
+//! The intermediate state is exposed as `Binary` to match the `BinaryType`
+//! buffer of Spark's `TypedImperativeAggregate` (see `state_fields`), which is
+//! the schema of the shuffle Exchange between the partial and final aggregate.
+//!
+//! A `GroupsAccumulator` fast path is provided for grouped aggregation, which
+//! is the common shape for `listagg`.
+
+use std::hash::Hash;
+use std::mem::size_of_val;
+use std::sync::Arc;
+
+use arrow::array::{Array, ArrayRef, BinaryArray, BooleanArray, StringArray};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion::common::cast::{as_binary_array, as_string_array};
+use datafusion::common::{internal_datafusion_err, not_impl_err, Result, ScalarValue};
+use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion::logical_expr::utils::format_state_name;
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, TypeSignature, Volatility,
+};
+use datafusion::physical_expr::expressions::Literal;
+
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkListAgg {
+ signature: Signature,
+}
+
+impl Default for SparkListAgg {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkListAgg {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::one_of(
+ vec![
+ TypeSignature::Exact(vec![DataType::Utf8, DataType::Utf8]),
+ TypeSignature::Exact(vec![DataType::Utf8, DataType::Null]),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+impl AggregateUDFImpl for SparkListAgg {
+ fn name(&self) -> &str {
+ "listagg"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Ok(DataType::Utf8)
+ }
+
+ fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
+ // The partial state is the group's concatenated string so far, exposed
+ // as `Binary`. Spark's `ListAgg` is a `TypedImperativeAggregate`, whose
+ // `aggBufferAttributes` is a single `BinaryType` column. Comet's partial
+ // `HashAggregate` output schema comes from those buffer attributes, so
+ // the shuffle Exchange between the partial and final aggregate carries
+ // `Binary`. The native state must match, even though partial and final
+ // always run in Comet (`CometListAgg` supports neither mixed-execution
+ // direction): the Exchange schema is Spark's, not DataFusion's.
+ Ok(vec![Field::new(
+ format_state_name(args.name, "listagg"),
+ DataType::Binary,
+ true,
+ )
+ .into()])
+ }
+
+ fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ListAggAccumulator::new(Self::extract_delimiter(
+ &acc_args,
+ )?)))
+ }
+
+ fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
+ true
+ }
+
+ fn create_groups_accumulator(
+ &self,
+ args: AccumulatorArgs,
+ ) -> Result<Box<dyn GroupsAccumulator>> {
+ Ok(Box::new(ListAggGroupsAccumulator::new(
+ Self::extract_delimiter(&args)?,
+ )))
+ }
+}
+
+impl SparkListAgg {
+ /// Read the delimiter string from the second argument. Spark's analyzer
+ /// guarantees a literal; a `NULL` literal is Spark's empty default.
+ fn extract_delimiter(args: &AccumulatorArgs) -> Result<String> {
+ let Some(lit) = (*args.exprs[1]).downcast_ref::<Literal>() else {
+ return not_impl_err!(
+ "listagg delimiter must be a literal; got {:?}",
+ args.exprs[1]
+ );
+ };
+ if lit.value().is_null() {
+ Ok(String::new())
+ } else if let Some(s) = lit.value().try_as_str() {
+ Ok(s.unwrap_or("").to_string())
+ } else {
+ not_impl_err!(
+ "listagg delimiter literal must be Utf8; got {:?}",
+ lit.value()
+ )
+ }
+ }
+}
+
+#[derive(Debug)]
+struct ListAggAccumulator {
+ delimiter: String,
+ accumulated: String,
+ has_value: bool,
+}
+
+impl ListAggAccumulator {
+ fn new(delimiter: String) -> Self {
+ Self {
+ delimiter,
+ accumulated: String::new(),
+ has_value: false,
+ }
+ }
+
+ #[inline]
+ fn append_values(&mut self, array: &StringArray) {
+ for value in array.iter().flatten() {
+ if self.has_value {
+ self.accumulated.push_str(&self.delimiter);
+ }
+ self.accumulated.push_str(value);
+ self.has_value = true;
+ }
+ }
+}
+
+impl Accumulator for ListAggAccumulator {
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ let array = values.first().ok_or_else(|| {
+ internal_datafusion_err!("listagg update_batch expected the values array")
+ })?;
+ self.append_values(as_string_array(array)?);
+ Ok(())
+ }
+
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+ let array = states.first().ok_or_else(|| {
+ internal_datafusion_err!("listagg merge_batch expected the state array")
+ })?;
+ // Partial state is a `Binary` array (see `state_fields`); each non-null
+ // entry is the UTF-8 bytes of another partition's already-joined group
+ // value.
+ let bin = as_binary_array(array)?;
+ for value in bin.iter().flatten() {
+ let s = std::str::from_utf8(value).map_err(|e| {
+ internal_datafusion_err!("listagg merge_batch got non-UTF-8 partial state: {e}")
+ })?;
+ if self.has_value {
+ self.accumulated.push_str(&self.delimiter);
+ }
+ self.accumulated.push_str(s);
+ self.has_value = true;
+ }
+ Ok(())
+ }
+
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ let value = if self.has_value {
+ ScalarValue::Binary(Some(std::mem::take(&mut self.accumulated).into_bytes()))
+ } else {
+ ScalarValue::Binary(None)
+ };
+ self.has_value = false;
+ Ok(vec![value])
+ }
+
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ if self.has_value {
+ Ok(ScalarValue::Utf8(Some(self.accumulated.clone())))
+ } else {
+ Ok(ScalarValue::Utf8(None))
+ }
+ }
+
+ fn size(&self) -> usize {
+ size_of_val(self) + self.delimiter.capacity() + self.accumulated.capacity()
+ }
+}
+
+/// Grouped fast path. `values[group]` is the group's concatenated string so
+/// far, or `None` if the group has seen no non-null input (emits `NULL`).
+/// Values are appended in arrival order, matching Spark's non-`WITHIN GROUP`
+/// concatenation order.
+#[derive(Debug)]
+struct ListAggGroupsAccumulator {
+ delimiter: String,
+ values: Vec<Option<String>>,
+ /// Running total of accumulated string bytes, for `size()`.
+ total_data_bytes: usize,
+}
+
+impl ListAggGroupsAccumulator {
+ fn new(delimiter: String) -> Self {
+ Self {
+ delimiter,
+ values: Vec::new(),
+ total_data_bytes: 0,
+ }
+ }
+
+ fn append(&mut self, group_idx: usize, value: &str) {
+ match &mut self.values[group_idx] {
+ Some(existing) => {
+ existing.push_str(&self.delimiter);
+ existing.push_str(value);
+ self.total_data_bytes += self.delimiter.len() + value.len();
+ }
+ slot @ None => {
+ *slot = Some(value.to_string());
+ self.total_data_bytes += value.len();
+ }
+ }
+ }
+
+ /// Drain the requested groups, updating the byte accounting. Both `evaluate`
+ /// (`Utf8` result) and `state` (`Binary` partial buffer) start here; they
+ /// differ only in the array type they build from the drained strings.
+ fn take_emitted(&mut self, emit_to: EmitTo) -> Vec<Option<String>> {
+ let emitted = emit_to.take_needed(&mut self.values);
+ let emitted_bytes: usize = emitted.iter().flatten().map(|s| s.len()).sum();
+ self.total_data_bytes -= emitted_bytes;
+ emitted
+ }
+}
+
+impl GroupsAccumulator for ListAggGroupsAccumulator {
+ fn update_batch(
+ &mut self,
+ values: &[ArrayRef],
+ group_indices: &[usize],
+ opt_filter: Option<&BooleanArray>,
+ total_num_groups: usize,
+ ) -> Result<()> {
+ self.values.resize(total_num_groups, None);
+ let array = as_string_array(&values[0])?;
+ for (i, &group_idx) in group_indices.iter().enumerate() {
+ if let Some(f) = opt_filter {
+ if !f.is_valid(i) || !f.value(i) {
+ continue;
+ }
+ }
+ if array.is_valid(i) {
+ self.append(group_idx, array.value(i));
+ }
+ }
+ Ok(())
+ }
+
+ fn evaluate(&mut self, emit_to: EmitTo) -> Result<ArrayRef> {
+ // Final result is `Utf8` (Spark `StringType`).
+ Ok(Arc::new(StringArray::from(self.take_emitted(emit_to))))
+ }
+
+ fn state(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
+ // Partial buffer is `Binary` (see `SparkListAgg::state_fields`).
+ let array: BinaryArray = self
+ .take_emitted(emit_to)
+ .into_iter()
+ .map(|opt| opt.map(String::into_bytes))
+ .collect();
+ Ok(vec![Arc::new(array)])
+ }
+
+ fn merge_batch(
+ &mut self,
+ values: &[ArrayRef],
+ group_indices: &[usize],
+ total_num_groups: usize,
+ ) -> Result<()> {
+ self.values.resize(total_num_groups, None);
+ // State is `Binary`: each non-null entry is another partition's joined
+ // group value as UTF-8 bytes.
+ let array = as_binary_array(&values[0])?;
+ for (i, &group_idx) in group_indices.iter().enumerate() {
+ if array.is_valid(i) {
+ let s = std::str::from_utf8(array.value(i)).map_err(|e| {
+ internal_datafusion_err!("listagg merge_batch got non-UTF-8 partial state: {e}")
+ })?;
+ self.append(group_idx, s);
+ }
+ }
+ Ok(())
+ }
+
+ fn convert_to_state(
+ &self,
+ _values: &[ArrayRef],
+ _opt_filter: Option<&BooleanArray>,
+ ) -> Result<Vec<ArrayRef>> {
+ not_impl_err!("Input batch conversion to state not implemented")
+ }
+
+ fn size(&self) -> usize {
+ self.total_data_bytes
+ + self.values.capacity() * std::mem::size_of::<Option<String>>()
+ + self.delimiter.capacity()
+ + size_of_val(self)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{BinaryArray, StringArray};
+
+ fn utf8(items: &[Option<&str>]) -> ArrayRef {
+ Arc::new(StringArray::from(items.to_vec()))
+ }
+
+ fn some(items: &[&str]) -> ArrayRef {
+ Arc::new(StringArray::from(items.to_vec()))
+ }
+
+ #[test]
+ fn joins_non_null_values_with_delimiter() -> Result<()> {
+ let mut acc = ListAggAccumulator::new(",".to_string());
+ acc.update_batch(&[some(&["a", "b", "c"])])?;
+ let ScalarValue::Utf8(Some(s)) = acc.evaluate()? else {
+ panic!("expected Utf8");
+ };
+ assert_eq!(s, "a,b,c");
+ Ok(())
+ }
+
+ #[test]
+ fn empty_delimiter_concatenates() -> Result<()> {
+ let mut acc = ListAggAccumulator::new(String::new());
+ acc.update_batch(&[some(&["a", "b", "c"])])?;
+ let ScalarValue::Utf8(Some(s)) = acc.evaluate()? else {
+ panic!("expected Utf8");
+ };
+ assert_eq!(s, "abc");
+ Ok(())
+ }
+
+ #[test]
+ fn skips_null_inputs() -> Result<()> {
+ let mut acc = ListAggAccumulator::new(",".to_string());
+ acc.update_batch(&[utf8(&[Some("a"), None, Some("b"), None])])?;
+ let ScalarValue::Utf8(Some(s)) = acc.evaluate()? else {
+ panic!("expected Utf8");
+ };
+ assert_eq!(s, "a,b");
+ Ok(())
+ }
+
+ #[test]
+ fn returns_null_on_all_null_or_empty_input() -> Result<()> {
+ let mut acc = ListAggAccumulator::new(",".to_string());
+ acc.update_batch(&[utf8(&[None, None])])?;
+ assert!(matches!(acc.evaluate()?, ScalarValue::Utf8(None)));
+
+ let mut empty = ListAggAccumulator::new(",".to_string());
+ assert!(matches!(empty.evaluate()?, ScalarValue::Utf8(None)));
+ Ok(())
+ }
+
+ #[test]
+ fn merge_state_across_partitions() -> Result<()> {
+ let mut a = ListAggAccumulator::new(",".to_string());
+ a.update_batch(&[some(&["a", "b"])])?;
+ let state_bytes = match a.state()?.remove(0) {
+ ScalarValue::Binary(Some(b)) => b,
+ other => panic!("unexpected state {other:?}"),
+ };
+
+ let mut b = ListAggAccumulator::new(",".to_string());
+ b.update_batch(&[some(&["c", "d"])])?;
+ let partial_state: ArrayRef =
+ Arc::new(BinaryArray::from(vec![Some(state_bytes.as_slice())]));
+ b.merge_batch(&[partial_state])?;
+
+ let ScalarValue::Utf8(Some(s)) = b.evaluate()? else {
+ panic!("expected Utf8");
+ };
+ // partition A's already-joined "a,b" is appended as one value.
+ assert_eq!(s, "c,d,a,b");
+ Ok(())
+ }
+
+ fn groups_result(acc: &mut ListAggGroupsAccumulator) -> Vec<Option<String>> {
+ let array = acc.evaluate(EmitTo::All).unwrap();
+ let strings = as_string_array(&array).unwrap();
+ strings
+ .iter()
+ .map(|opt| opt.map(|s| s.to_string()))
+ .collect()
+ }
+
+ #[test]
+ fn groups_accumulator_joins_per_group() -> Result<()> {
+ let mut acc = ListAggGroupsAccumulator::new(",".to_string());
+ // group 0: "a", "c"; group 1: "b"; group 2 (index 2) never seen -> NULL.
+ acc.update_batch(
+ &[utf8(&[Some("a"), Some("b"), None, Some("c")])],
+ &[0, 1, 1, 0],
+ None,
+ 3,
+ )?;
+ assert_eq!(
+ groups_result(&mut acc),
+ vec![Some("a,c".to_string()), Some("b".to_string()), None]
+ );
+ Ok(())
+ }
+
+ #[test]
+ fn groups_accumulator_merges_partial_state() -> Result<()> {
+ // Two partitions each produce partial state, then a final accumulator
+ // merges them in arrival order.
+ let mut p0 = ListAggGroupsAccumulator::new(",".to_string());
+ p0.update_batch(&[some(&["a", "b"])], &[0, 0], None, 1)?;
+ let s0 = p0.state(EmitTo::All)?.remove(0);
+
+ let mut p1 = ListAggGroupsAccumulator::new(",".to_string());
+ p1.update_batch(&[some(&["c"])], &[0], None, 1)?;
+ let s1 = p1.state(EmitTo::All)?.remove(0);
+
+ let mut fin = ListAggGroupsAccumulator::new(",".to_string());
+ fin.merge_batch(&[s0], &[0], 1)?;
+ fin.merge_batch(&[s1], &[0], 1)?;
+ assert_eq!(groups_result(&mut fin), vec![Some("a,b,c".to_string())]);
+ Ok(())
+ }
+}
diff --git a/native/spark-expr/src/agg_funcs/mod.rs b/native/spark-expr/src/agg_funcs/mod.rs
index fcbe465c78..cfe0656519 100644
--- a/native/spark-expr/src/agg_funcs/mod.rs
+++ b/native/spark-expr/src/agg_funcs/mod.rs
@@ -23,6 +23,7 @@ mod correlation;
mod covariance;
mod hll_plus_plus;
mod hll_plus_plus_const;
+mod list_agg;
mod max_min_by;
mod mode;
mod percentile;
@@ -41,6 +42,7 @@ pub use collect::{CometCollectList, CometCollectSet};
pub use correlation::Correlation;
pub use covariance::Covariance;
pub use hll_plus_plus::{hllpp_precision, HllPlusPlus};
+pub use list_agg::SparkListAgg;
pub use max_min_by::MaxMinBy;
pub use mode::Mode;
pub use percentile::SparkPercentile;
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index 43d65fd418..aa4399c3b3 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -409,38 +409,41 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
/**
* Mapping of Spark aggregate expression class to Comet expression handler.
*/
- val aggrSerdeMap: Map[Class[_], CometAggregateExpressionSerde[_]] = Map(
- classOf[ApproximatePercentile] -> CometApproxPercentile,
- classOf[HyperLogLogPlusPlus] -> CometApproxCountDistinct,
- classOf[Average] -> CometAverage,
- classOf[BitAndAgg] -> CometBitAndAgg,
- classOf[BitOrAgg] -> CometBitOrAgg,
- classOf[BitXorAgg] -> CometBitXOrAgg,
- classOf[BloomFilterAggregate] -> CometBloomFilterAggregate,
- classOf[CollectList] -> CometCollectList,
- classOf[CollectSet] -> CometCollectSet,
- classOf[Corr] -> CometCorr,
- classOf[Count] -> CometCount,
- classOf[CovPopulation] -> CometCovPopulation,
- classOf[CovSample] -> CometCovSample,
- classOf[First] -> CometFirst,
- classOf[Last] -> CometLast,
- classOf[Max] -> CometMax,
- classOf[MaxBy] -> CometMaxBy,
- classOf[Min] -> CometMin,
- classOf[MinBy] -> CometMinBy,
- classOf[Mode] -> CometMode,
- classOf[Percentile] -> CometPercentile,
- classOf[RegrIntercept] -> CometRegrIntercept,
- classOf[RegrR2] -> CometRegrR2,
- classOf[RegrReplacement] -> CometRegrReplacement,
- classOf[RegrSlope] -> CometRegrSlope,
- classOf[RegrSXY] -> CometRegrSXY,
- classOf[StddevPop] -> CometStddevPop,
- classOf[StddevSamp] -> CometStddevSamp,
- classOf[Sum] -> CometSum,
- classOf[VariancePop] -> CometVariancePop,
- classOf[VarianceSamp] -> CometVarianceSamp)
+ val aggrSerdeMap: Map[Class[_], CometAggregateExpressionSerde[_]] = {
+ val base: Map[Class[_], CometAggregateExpressionSerde[_]] = Map(
+ classOf[ApproximatePercentile] -> CometApproxPercentile,
+ classOf[HyperLogLogPlusPlus] -> CometApproxCountDistinct,
+ classOf[Average] -> CometAverage,
+ classOf[BitAndAgg] -> CometBitAndAgg,
+ classOf[BitOrAgg] -> CometBitOrAgg,
+ classOf[BitXorAgg] -> CometBitXOrAgg,
+ classOf[BloomFilterAggregate] -> CometBloomFilterAggregate,
+ classOf[CollectList] -> CometCollectList,
+ classOf[CollectSet] -> CometCollectSet,
+ classOf[Corr] -> CometCorr,
+ classOf[Count] -> CometCount,
+ classOf[CovPopulation] -> CometCovPopulation,
+ classOf[CovSample] -> CometCovSample,
+ classOf[First] -> CometFirst,
+ classOf[Last] -> CometLast,
+ classOf[Max] -> CometMax,
+ classOf[MaxBy] -> CometMaxBy,
+ classOf[Min] -> CometMin,
+ classOf[MinBy] -> CometMinBy,
+ classOf[Mode] -> CometMode,
+ classOf[Percentile] -> CometPercentile,
+ classOf[RegrIntercept] -> CometRegrIntercept,
+ classOf[RegrR2] -> CometRegrR2,
+ classOf[RegrReplacement] -> CometRegrReplacement,
+ classOf[RegrSlope] -> CometRegrSlope,
+ classOf[RegrSXY] -> CometRegrSXY,
+ classOf[StddevPop] -> CometStddevPop,
+ classOf[StddevSamp] -> CometStddevSamp,
+ classOf[Sum] -> CometSum,
+ classOf[VariancePop] -> CometVariancePop,
+ classOf[VarianceSamp] -> CometVarianceSamp)
+ base ++ sparkVersionSpecificAggregates
+ }
/**
* Returns true if Spark can consume all the intermediate buffers produced by Comet. Used when a
diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala
index 594c85e02f..6ccc247046 100644
--- a/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala
+++ b/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala
@@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate.Sum
import org.apache.comet.expressions.CometEvalMode
-import org.apache.comet.serde.{CometEncode, CometExpressionSerde, CometStringDecode}
+import org.apache.comet.serde.{CometAggregateExpressionSerde, CometEncode, CometExpressionSerde, CometStringDecode}
import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr}
/**
@@ -44,6 +44,8 @@ trait CometExprShim {
Map.empty
def sparkVersionSpecificMapExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] =
Map.empty
+ def sparkVersionSpecificAggregates: Map[Class[_], CometAggregateExpressionSerde[_]] =
+ Map.empty
def sparkVersionSpecificExprToProtoInternal(
expr: Expression,
diff --git a/spark/src/main/spark-3.5/org/apache/comet/shims/CometExprShim.scala b/spark/src/main/spark-3.5/org/apache/comet/shims/CometExprShim.scala
index 5cd3ae549d..e9383ae280 100644
--- a/spark/src/main/spark-3.5/org/apache/comet/shims/CometExprShim.scala
+++ b/spark/src/main/spark-3.5/org/apache/comet/shims/CometExprShim.scala
@@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate.Sum
import org.apache.comet.expressions.CometEvalMode
-import org.apache.comet.serde.{CometEncode, CometExpressionSerde, CometStringDecode, CometToPrettyString}
+import org.apache.comet.serde.{CometAggregateExpressionSerde, CometEncode, CometExpressionSerde, CometStringDecode, CometToPrettyString}
import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr}
/**
@@ -44,6 +44,8 @@ trait CometExprShim {
Map(classOf[ToPrettyString] -> CometToPrettyString)
def sparkVersionSpecificMapExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] =
Map.empty
+ def sparkVersionSpecificAggregates: Map[Class[_], CometAggregateExpressionSerde[_]] =
+ Map.empty
def sparkVersionSpecificExprToProtoInternal(
expr: Expression,
diff --git a/spark/src/main/spark-4.x/org/apache/comet/serde/CometListAgg.scala b/spark/src/main/spark-4.x/org/apache/comet/serde/CometListAgg.scala
new file mode 100644
index 0000000000..e145dc9b18
--- /dev/null
+++ b/spark/src/main/spark-4.x/org/apache/comet/serde/CometListAgg.scala
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.serde
+
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, ListAgg}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{NullType, StringType}
+
+import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
+import org.apache.comet.serde.QueryPlanSerde.{exprToProto, hasNonDefaultStringCollation}
+
+/**
+ * Spark 4.0+ `LISTAGG` / `STRING_AGG`.
+ *
+ * Comet only supports the simple form: a `StringType` child with a literal delimiter and no
+ * `WITHIN GROUP (ORDER BY ...)` clause. DISTINCT falls back to Spark because Comet rejects
+ * multi-column distinct aggregates in `aggExprToProto`, so the native side never sees it.
+ */
+object CometListAgg extends CometAggregateExpressionSerde[ListAgg] {
+
+ private val binaryChildReason = "`BinaryType` inputs are not supported."
+ private val withinGroupReason = "`WITHIN GROUP (ORDER BY ...)` is not supported."
+ private val nonFoldableDelimiterReason = "Non-literal delimiters are not supported."
+ private val collationReason = "Non-default string collations are not supported."
+
+ // Note: DISTINCT is not listed here. It never reaches this serde: Comet's aggregate
+ // dispatcher rejects multi-column distinct aggregates earlier in `aggExprToProto`, which
+ // emits its own fallback reason.
+ override def getUnsupportedReasons(): Seq[String] =
+ Seq(binaryChildReason, withinGroupReason, nonFoldableDelimiterReason, collationReason)
+
+ override def getSupportLevel(expr: ListAgg): SupportLevel = {
+ // Spark's analyzer already enforces `delimiter.foldable`, so this only ever rejects
+ // non-string / non-null delimiter types.
+ expr.child.dataType match {
+ case _: StringType if hasNonDefaultStringCollation(expr.child.dataType) =>
+ Unsupported(Some(collationReason))
+ case _: StringType =>
+ expr.delimiter.dataType match {
+ case _: StringType if hasNonDefaultStringCollation(expr.delimiter.dataType) =>
+ Unsupported(Some(collationReason))
+ case _: StringType | _: NullType =>
+ if (!expr.delimiter.foldable) {
+ Unsupported(Some(nonFoldableDelimiterReason))
+ } else if (expr.orderExpressions.nonEmpty) {
+ Unsupported(Some(withinGroupReason))
+ } else {
+ Compatible()
+ }
+ case other =>
+ Unsupported(Some(s"Unsupported delimiter data type: $other"))
+ }
+ case other =>
+ Unsupported(Some(s"Unsupported child data type: $other"))
+ }
+ }
+
+ override def convert(
+ aggExpr: AggregateExpression,
+ expr: ListAgg,
+ inputs: Seq[Attribute],
+ binding: Boolean,
+ conf: SQLConf): Option[ExprOuterClass.AggExpr] = {
+ val childExpr = exprToProto(expr.child, inputs, binding)
+ val delimiterExpr = exprToProto(expr.delimiter, inputs, binding)
+
+ if (childExpr.isDefined && delimiterExpr.isDefined) {
+ val builder = ExprOuterClass.ListAgg.newBuilder()
+ builder.setChild(childExpr.get)
+ builder.setDelimiter(delimiterExpr.get)
+ Some(
+ ExprOuterClass.AggExpr
+ .newBuilder()
+ .setListAgg(builder)
+ .build())
+ } else {
+ withFallbackReason(aggExpr, "Child expression or data type not supported")
+ None
+ }
+ }
+}
diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
index 89663800ed..46cd0d1716 100644
--- a/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
+++ b/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
@@ -20,13 +20,14 @@
package org.apache.comet.shims
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.ListAgg
import org.apache.spark.sql.catalyst.expressions.json.{JsonExpressionUtils, StructsToJsonEvaluator}
import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke}
import org.apache.spark.sql.catalyst.expressions.url.ParseUrlEvaluator
import org.apache.comet.CometExplainInfo
import org.apache.comet.expressions.CometEvalMode
-import org.apache.comet.serde.{CometExpressionSerde, CometMapSort, CometRandStr, CometToPrettyString}
+import org.apache.comet.serde.{CometAggregateExpressionSerde, CometExpressionSerde, CometListAgg, CometMapSort, CometRandStr, CometToPrettyString}
import org.apache.comet.serde.ExprOuterClass.Expr
import org.apache.comet.serde.QueryPlanSerde.exprToProtoInternal
@@ -48,6 +49,8 @@ trait Spark4xCometExprShim extends CometExprShim4x {
Map(classOf[ToPrettyString] -> CometToPrettyString)
def sparkVersionSpecificMapExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] =
Map(classOf[MapSort] -> CometMapSort)
+ def sparkVersionSpecificAggregates: Map[Class[_], CometAggregateExpressionSerde[_]] =
+ Map(classOf[ListAgg] -> CometListAgg)
def sparkVersionSpecificExprToProtoInternal(
expr: Expression,
diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/listagg.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/listagg.sql
new file mode 100644
index 0000000000..3baa443608
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/aggregate/listagg.sql
@@ -0,0 +1,192 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- listagg / string_agg is available starting in Spark 4.0.
+-- MinSparkVersion: 4.0
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+-- ============================================================
+-- Setup
+-- ============================================================
+
+statement
+CREATE TABLE la_src(v string, grp string) USING parquet
+
+statement
+INSERT INTO la_src VALUES
+ ('a', 'g1'), ('b', 'g1'), ('c', 'g1'),
+ ('x', 'g2'), (NULL, 'g2'), ('y', 'g2'),
+ (NULL, 'g3'), (NULL, 'g3'),
+ ('', 'g4'), ('z', 'g4')
+
+statement
+CREATE TABLE la_empty(v string) USING parquet
+
+statement
+CREATE TABLE la_dupes(v string, grp string) USING parquet
+
+statement
+INSERT INTO la_dupes VALUES
+ ('a', 'g1'), ('a', 'g1'), ('b', 'g1'),
+ ('c', 'g2'), ('c', 'g2'), ('c', 'g2')
+
+statement
+CREATE TABLE la_utf8(v string, grp string) USING parquet
+
+-- Multibyte UTF-8: `cafΓ©` (composed Γ© U+00E9), `naΓ―ve` (composed Γ― U+00EF),
+-- `ζ₯ζ¬θͺ` (three CJK codepoints), and an emoji sequence.
+statement
+INSERT INTO la_utf8 VALUES
+ ('cafΓ©', 'g1'), ('naΓ―ve', 'g1'), ('ζ₯ζ¬θͺ', 'g1'),
+ ('νκΈ', 'g2'), ('βοΈ', 'g2'), (NULL, 'g2')
+
+statement
+CREATE TABLE la_bin(v binary, grp string) USING parquet
+
+statement
+INSERT INTO la_bin VALUES
+ (X'DEAD', 'g1'), (X'BEEF', 'g1'), (NULL, 'g1'),
+ (X'CAFE', 'g2')
+
+-- Identical values per group so the concatenation is order-independent and the
+-- multi-partition merge below stays deterministic.
+statement
+CREATE TABLE la_mp(v string, grp string) USING parquet
+
+statement
+INSERT INTO la_mp VALUES
+ ('a', 'g1'), ('a', 'g1'), ('a', 'g1'),
+ ('b', 'g2'), ('b', 'g2')
+
+-- ============================================================
+-- Basic: literal delimiter, sort the group so results are
+-- deterministic across shuffles.
+-- ============================================================
+
+query
+SELECT grp, listagg(v, ',') FROM (SELECT * FROM la_src ORDER BY grp, v) GROUP BY grp ORDER BY grp
+
+-- Alias `string_agg` should route through the same expression class.
+query
+SELECT grp, string_agg(v, ',') FROM (SELECT * FROM la_src ORDER BY grp, v) GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- No delimiter (defaults to empty string / NULL delimiter)
+-- ============================================================
+
+query
+SELECT grp, listagg(v) FROM (SELECT * FROM la_src ORDER BY grp, v) GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- All-NULL group returns NULL; empty table returns NULL.
+-- ============================================================
+
+query
+SELECT listagg(v, ',') FROM la_empty
+
+query
+SELECT grp, listagg(v, '-') FROM (SELECT * FROM la_src WHERE grp = 'g3' ORDER BY v) GROUP BY grp
+
+-- ============================================================
+-- DISTINCT falls back to Spark: Comet rejects multi-column
+-- distinct aggregates (listagg has two children: value and
+-- delimiter), so `listagg(DISTINCT v, ',')` is not run natively.
+-- ============================================================
+
+query expect_fallback(Multi-column distinct aggregate not supported)
+SELECT grp, listagg(DISTINCT v, ',') FROM (SELECT * FROM la_dupes ORDER BY grp, v) GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Global aggregate (no GROUP BY)
+-- ============================================================
+
+query
+SELECT listagg(v, '|') FROM (SELECT * FROM la_src WHERE grp = 'g1' ORDER BY v)
+
+-- ============================================================
+-- Multi-char delimiter and empty-string delimiter
+-- ============================================================
+
+query
+SELECT grp, listagg(v, ' -> ') FROM (SELECT * FROM la_src WHERE grp = 'g1' ORDER BY v) GROUP BY grp
+
+query
+SELECT grp, listagg(v, '') FROM (SELECT * FROM la_src WHERE grp = 'g1' ORDER BY v) GROUP BY grp
+
+-- ============================================================
+-- Empty-string values inside the group
+-- ============================================================
+
+query
+SELECT grp, listagg(v, ',') FROM (SELECT * FROM la_src WHERE grp = 'g4' ORDER BY v) GROUP BY grp
+
+-- ============================================================
+-- Multibyte UTF-8 values (composed accents, CJK, emoji)
+-- ============================================================
+
+query
+SELECT grp, listagg(v, ',') FROM (SELECT * FROM la_utf8 ORDER BY grp, v) GROUP BY grp ORDER BY grp
+
+-- Multi-byte delimiter
+query
+SELECT grp, listagg(v, 'Β·') FROM (SELECT * FROM la_utf8 ORDER BY grp, v) GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- WITHIN GROUP (ORDER BY ...) is not implemented natively;
+-- both ascending and descending must fall back to Spark.
+-- ============================================================
+
+query expect_fallback(`WITHIN GROUP (ORDER BY ...)` is not supported)
+SELECT grp, listagg(v, ',') WITHIN GROUP (ORDER BY v) FROM la_src GROUP BY grp ORDER BY grp
+
+query expect_fallback(`WITHIN GROUP (ORDER BY ...)` is not supported)
+SELECT grp, listagg(v, ',') WITHIN GROUP (ORDER BY v DESC) FROM la_src GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- BinaryType inputs are not supported natively; must fall back.
+-- ============================================================
+
+query expect_fallback(Unsupported child data type: BinaryType)
+SELECT grp, listagg(v) FROM la_bin GROUP BY grp ORDER BY grp
+
+query expect_fallback(Unsupported child data type: BinaryType)
+SELECT grp, listagg(v, X'42') FROM la_bin GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Non-default string collations are not supported natively.
+-- The delimiter must share the child's collation or the analyzer
+-- rejects the call, so it is collated explicitly.
+-- ============================================================
+
+statement
+CREATE TABLE la_coll(v string COLLATE UTF8_LCASE, grp string) USING parquet
+
+statement
+INSERT INTO la_coll VALUES ('a', 'g1'), ('B', 'g1')
+
+query expect_fallback(Non-default string collations are not supported)
+SELECT grp, listagg(v, collate(',', 'UTF8_LCASE')) FROM la_coll GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Multi-partition merge: force a round-robin repartition so each
+-- group is split across partitions and merged in the final
+-- aggregate. Values are identical within each group, so the
+-- result is deterministic regardless of merge order.
+-- ============================================================
+
+query
+SELECT grp, listagg(v, ',') FROM (SELECT /*+ REPARTITION(4) */ * FROM la_mp) GROUP BY grp ORDER BY grp
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org