You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/06/24 16:26:53 UTC
[PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
andygrove opened a new pull request, #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720
## Which issue does this PR close?
Closes #2524.
## Rationale for this change
`collect_list` and its alias `array_agg` are common aggregate functions that previously fell back to Spark, breaking native execution for many real workloads (notably plans that group rows into arrays before further processing). Adding native support keeps these queries on the Comet path.
## What changes are included in this PR?
This change was scaffolded with the `implement-comet-expression` skill.
- Add `CollectList` message to `expr.proto` and a new `collectList = 18` arm in the `AggExpr` oneof.
- Add `CometCollectList` serde in `aggregates.scala` and register `classOf[CollectList] -> CometCollectList` in `QueryPlanSerde`.
- Wire the native side to `datafusion_spark::function::aggregate::collect::SparkCollectList`, the upstream Spark-compatible accumulator (Spark 3.4 through 4.1 use `ignore_nulls = true` semantics that match it). No Comet-local Rust function is added.
- Extend the Partial-mode `adjustOutputForNativeState` fix in `operators.scala` to cover `CollectList` (same pattern already used for `CollectSet`: native produces `ArrayType(elementType)` while Spark declares the buffer as `BinaryType`).
- Mark `collect_list` and `array_agg` as supported in the user-guide expression page.
- Add an audit entry under `docs/source/contributor-guide/expression-audits/agg_funcs.md` covering Spark 3.4.3, 3.5.8, 4.0.1, 4.1.1.
## How are these changes tested?
- New SQL fixture `spark/src/test/resources/sql-tests/expressions/aggregate/collect_list.sql` exercises ~30 queries across types (boolean, byte/short/int/bigint, float/double including NaN/Inf/-0, string, binary, decimal up to (38,0), date, timestamp, struct, nested array), GROUP BY, NULLs, all-NULL groups, empty tables, single-row, mixed aggregates, multiple `collect_list` columns, DISTINCT, HAVING, the `array_agg` alias, INT/BIGINT boundary values, and an SPARK-17641 null-filter regression. The fixture runs under `ConfigMatrix: parquet.enable.dictionary=false,true`, so each query executes twice.
- All 21 aggregate `CometSqlFileTestSuite` tests pass (`./mvnw test -Dsuites="org.apache.comet.CometSqlFileTestSuite expressions/aggregate" -Dtest=none -Pspark-3.5`), confirming no regression to `collect_set` or other aggregates.
- `cargo clippy --all-targets --workspace -- -D warnings` is clean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3604312544
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -858,13 +858,19 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
" `spark.comet.expression.CollectSet.allowIncompatible=true` is set.")
override def getSupportLevel(expr: CollectSet): SupportLevel = {
- SupportLevel
- .strictFloatingPointReason(
- expr.children.head.dataType,
- "collect_set on floating-point types " +
- "(Comet deduplicates NaN values while Spark treats each NaN as distinct)")
- .map(reason => Incompatible(Some(reason)))
- .getOrElse(Compatible())
+ // The native path always drops null inputs. Spark 4.2 adds `RESPECT NULLS`
+ // (`ignoreNulls = false`), which keeps nulls, so fall back there.
+ if (!CometCollectShim.ignoreNulls(expr)) {
Review Comment:
this code is confusing
```
spark.sql("SELECT collect_list(c) IGNORE NULLS FROM VALUES (1),(NULL) AS t(c);").show(false)
org.apache.spark.sql.AnalysisException: [INVALID_SQL_SYNTAX.FUNCTION_WITH_UNSUPPORTED_SYNTAX] Invalid SQL syntax: The function `collect_list` does not support IGNORE NULLS.
```
it is always dead code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3604318907
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -740,6 +740,38 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
}
}
+object CometCollectList extends CometAggregateExpressionSerde[CollectList] {
+
+ override def convert(
+ aggExpr: AggregateExpression,
+ expr: CollectList,
+ inputs: Seq[Attribute],
+ binding: Boolean,
+ conf: SQLConf): Option[ExprOuterClass.AggExpr] = {
+ val child = expr.children.head
Review Comment:
ignore nulls is not supported by spark for those functions, Comet shouldn't be handling it as this is misleading
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3554833046
##########
dev/diffs/3.5.8.diff:
##########
@@ -266,7 +266,21 @@ index 6f3090d8908..a0e9309888d 100644
}
// test case for ObjectHashAggregate and SortAggregate
-@@ -793,7 +796,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -771,7 +774,12 @@ class DataFrameAggregateSuite extends QueryTest
+ }
+
+ test("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") {
+- withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") {
++ withSQLConf(
++ SQLConf.USE_OBJECT_HASH_AGG.key -> "true",
++ // Comet runs collect_list/collect_set natively (CometHashAggregateExec), so the
++ // ObjectHashAggregateExec this test asserts on is no longer present. Disable Comet
++ // to preserve the original plan-shape assertions.
++ "spark.comet.enabled" -> "false") {
Review Comment:
Agreed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3582281900
##########
dev/diffs/3.5.8.diff:
##########
@@ -266,7 +266,21 @@ index 6f3090d8908..a0e9309888d 100644
}
// test case for ObjectHashAggregate and SortAggregate
-@@ -793,7 +796,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -771,7 +774,12 @@ class DataFrameAggregateSuite extends QueryTest
+ }
+
+ test("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") {
+- withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") {
++ withSQLConf(
++ SQLConf.USE_OBJECT_HASH_AGG.key -> "true",
++ // Comet runs collect_list/collect_set natively (CometHashAggregateExec), so the
++ // ObjectHashAggregateExec this test asserts on is no longer present. Disable Comet
++ // to preserve the original plan-shape assertions.
++ "spark.comet.enabled" -> "false") {
Review Comment:
this is fixed now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3604073509
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1643,6 +1643,21 @@ trait CometBaseAggregate {
}
}
+ // CollectList/CollectSet declare their buffer as BinaryType in Spark but produce a native
Review Comment:
Dropped. You are right, the general missingCometProducer + aggsNotSupportingMixedExecution guard just above already returns None for the same case, so this block was unreachable.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -871,6 +871,24 @@ case class CometExecRule(session: SparkSession)
}
}
}
+
+ // CollectList/CollectSet round-trip an ArrayType buffer that Spark declares as BinaryType.
Review Comment:
Added a comment on the block explaining why it is separate: the tagging block above only fires when the Final cannot convert, but canAggregateBeConverted skips the child-native check, so an all-native distinct collect_list Final converts and slips past. This block catches that case.
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -439,6 +440,22 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
}
}
+ /**
+ * Returns true if any aggregate function produces a native intermediate buffer whose Arrow type
+ * (e.g. ArrayType for CollectList/CollectSet) differs from the BinaryType that Spark declares
Review Comment:
Narrowed the doc comment to describe what the code actually matches (CollectList/CollectSet) and noted that Percentile has the same buffer shape but is intentionally not matched here since it already passes through the general mixed-execution guard.
##########
spark/src/main/spark-4.2/org/apache/comet/shims/CometCollectShim.scala:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.shims
+
+import org.apache.spark.sql.catalyst.expressions.aggregate.{CollectList, CollectSet}
+
+/**
+ * Shim for the `ignoreNulls` flag on `CollectList` / `CollectSet`. Spark 4.2 added the field
+ * (`ignoreNulls = false` via `RESPECT NULLS`), so this shim reports the actual value. The serde
+ * falls back to Spark when it is `false`, since the native path always drops nulls.
+ */
+object CometCollectShim {
Review Comment:
Moved the identical spark-3.4/spark-3.5 copies to spark-3.x. spark-4.x is used by spark-4.0/4.1/4.2 too, and 4.2 needs the different shim, so I could not consolidate 4.0/4.1 there — happy to add a fresh shared dir if you would prefer that, but it seemed like more build machinery than the reduction was worth.
##########
spark/src/test/resources/sql-tests/expressions/aggregate/collect_list.sql:
##########
@@ -0,0 +1,405 @@
+-- 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.
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+-- collect_list result order is non-deterministic across partitions, so
+-- every query wraps the result in sort_array to make comparisons stable.
+
+-- ============================================================
+-- Setup: tables
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_int(i int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_int VALUES
+ (1, 'a'), (2, 'a'), (1, 'a'), (3, 'a'),
+ (4, 'b'), (4, 'b'), (NULL, 'b'), (5, 'b'),
+ (NULL, 'c'), (NULL, 'c')
+
+statement
+CREATE TABLE cl_src_nulls(val int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_nulls VALUES
Review Comment:
Added `collect_list FILTER (WHERE i > 1)` per-group and a map-input case (`collect_list(m)` on a `map<string, int>` table with two entries and a null row). The map assertion is a `spark_answer_only` size-only check since sort_array cannot order a MapType.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3554800962
##########
dev/diffs/3.5.8.diff:
##########
@@ -266,7 +266,21 @@ index 6f3090d8908..a0e9309888d 100644
}
// test case for ObjectHashAggregate and SortAggregate
-@@ -793,7 +796,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -771,7 +774,12 @@ class DataFrameAggregateSuite extends QueryTest
+ }
+
+ test("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") {
+- withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") {
++ withSQLConf(
++ SQLConf.USE_OBJECT_HASH_AGG.key -> "true",
++ // Comet runs collect_list/collect_set natively (CometHashAggregateExec), so the
++ // ObjectHashAggregateExec this test asserts on is no longer present. Disable Comet
++ // to preserve the original plan-shape assertions.
++ "spark.comet.enabled" -> "false") {
Review Comment:
So we're disabling Comet? That doesn't seem like the wrong approach.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3554839721
##########
dev/diffs/3.5.8.diff:
##########
@@ -266,7 +266,21 @@ index 6f3090d8908..a0e9309888d 100644
}
// test case for ObjectHashAggregate and SortAggregate
-@@ -793,7 +796,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -771,7 +774,12 @@ class DataFrameAggregateSuite extends QueryTest
+ }
+
+ test("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") {
+- withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") {
++ withSQLConf(
++ SQLConf.USE_OBJECT_HASH_AGG.key -> "true",
++ // Comet runs collect_list/collect_set natively (CometHashAggregateExec), so the
++ // ObjectHashAggregateExec this test asserts on is no longer present. Disable Comet
++ // to preserve the original plan-shape assertions.
++ "spark.comet.enabled" -> "false") {
Review Comment:
Moving to draft until this is resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#issuecomment-5071730068
Agent found only Medium severity issues
## Comment 1 — docs/source/contributor-guide/expression-audits/agg_funcs.md
This PR modifies `CometCollectSet` semantically (new RESPECT NULLS branch via `CometCollectShim.ignoreNulls(CollectSet)`, name-matched in `hasNativeArrayBufferAgg`, and referenced twice from the new `##
collect_list` section) but the file has no `collect_set` entry at all. Please add one alongside `## collect_list` so future auditors don't diverge the two. Suggested content, matching the existing template:
```markdown
## collect_set
- Spark 3.4.3 (audited 2026-06-24): `CollectSet` extends `Collect[mutable.HashSet[Any]]`, returns `ArrayType(child.dataType, containsNull = false)`, ignores NULL inputs, uses value-equality for dedup so `NaN
!= NaN` is kept as distinct, and yields an empty array as `defaultResult`. `nullable = false`. Registered under `collect_set` only (no alias).
- Spark 3.5.8 (audited 2026-06-24): identical to 3.4.3.
- Spark 4.0.1 (audited 2026-06-24): adds `with UnaryLike[Expression]`; no behavior change.
- Spark 4.1.1 (audited 2026-06-24): identical to 4.0.1.
- Comet implementation: native side delegates to `datafusion_spark::function::aggregate::collect::SparkCollectSet` (see the collect_list entry for the same buffer-shape mismatch and `containsNull`
normalization). Because Spark treats each NaN as distinct while the native path dedups by IEEE-754 equality (`NaN == NaN`), `CometCollectSet.getSupportLevel` reports `Incompatible` on `float`/`double` inputs
via `SupportLevel.strictFloatingPointReason`, gated on `spark.comet.expression.CollectSet.allowIncompatible=true`.
- Spark 4.2 (preview): shares the `ignoreNulls` field with `CollectList`; `RESPECT NULLS` triggers fallback to Spark (see the collect_list entry).
```
Then in the `## collect_list` entry, replace *"same pattern already in use for collect_set"* with a link to the new `## collect_set` section.
## Comment 2 — spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
`hasNativeArrayBufferAgg` + `tagUnsafePartialAggregates` exist specifically to make the split-execution case safe, but neither new test toggles `COMET_ENABLE_FINAL_HASH_AGGREGATE` /
`COMET_ENABLE_PARTIAL_HASH_AGGREGATE`. Please mirror the `"mixed engine sum/avg"` tests at lines 211/224 so the cascade is regression-covered:
```scala
test("mixed engine collect_list: Comet partial + Spark final matches Spark") {
import org.apache.spark.sql.functions.{collect_list, sort_array}
val data = (0 until 100).map(i => (i, s"v$i", i % 7))
withParquetTable(data, "tbl") {
withSQLConf(
CometConf.COMET_ENABLE_FINAL_HASH_AGGREGATE.key -> "false",
CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm") {
checkSparkAnswer(
"SELECT _3, sort_array(collect_list(_1)), sort_array(collect_list(_2)) " +
"FROM tbl GROUP BY _3")
}
}
}
test("mixed engine collect_list: Spark partial + Comet final matches Spark") {
// Same body, toggling COMET_ENABLE_PARTIAL_HASH_AGGREGATE=false.
// This is the case that must fall back per the adjustOutputForNativeState contract —
// Comet cannot read Spark's BinaryType buffer.
}
```
The second variant is where the fallback machinery actually earns its keep; if it silently ran natively, `adjustOutputForNativeState` would misinterpret Spark's Binary state as an Array.
## Comment 3 — spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala (in the new "collect_list/collect_set combined with distinct aggregate falls back safely" test)
`checkSparkAnswer` doesn't verify that anything ran natively — a future change that widens the `hasNativeArrayBufferAgg` guard to tag partials for unrelated aggregates (or fully falls back for another
reason) would still pass here silently. Please assert the exact fallback reason so this test actually protects the #4724 guard:
```scala
val reason =
"Partial aggregate disabled: part of a multi-stage CollectList/CollectSet " +
"aggregate whose intermediate buffer cannot round-trip in Comet (issue #4724)"
checkSparkAnswerAndFallbackReason(
df.groupBy(col("x"))
.agg(count_distinct(col("y")), sort_array(collect_list(col("z")))),
reason)
```
Apply the same to the `collect_set` variant and to both branches (LocalTableScan + Parquet). `checkSparkAnswerAndFallbackReason` is defined at `CometTestBase.scala:296`.
## Comment 4 — spark/src/test/resources/sql-tests/expressions/aggregate/collect_list.sql
The file exercises `timestamp` but not `timestamp_ntz`, and no interval types — yet `TimestampNTZType` is in `QueryPlanSerde.supportedDataType`, so `collect_list(ts_ntz_col)` flows through the native
`SparkCollectList` accumulator (an untested native path). Year-month / day-time intervals only exercise the Spark fallback but are required by the project's *"every datatype Spark accepts, NULLs in every
column"* rule. Please append:
```sql
-- ============================================================
-- Timestamp NTZ (with NULLs) — exercises the native path
-- ============================================================
statement
CREATE TABLE cl_src_ts_ntz(v timestamp_ntz, grp string) USING parquet
statement
INSERT INTO cl_src_ts_ntz VALUES
(TIMESTAMP_NTZ '2024-01-01 00:00:00', 'a'),
(TIMESTAMP_NTZ '2024-06-15 12:30:00', 'a'),
(TIMESTAMP_NTZ '2024-01-01 00:00:00', 'a'),
(NULL, 'a'),
(TIMESTAMP_NTZ '1970-01-01 00:00:00', 'b'),
(NULL, 'b')
query
SELECT grp, sort_array(collect_list(v)) FROM cl_src_ts_ntz GROUP BY grp ORDER BY grp
-- ============================================================
-- Year-Month / Day-Time intervals (with NULLs) — falls back to Spark
-- ============================================================
statement
CREATE TABLE cl_src_intervals(ym interval year to month, dt interval day to second, grp string) USING parquet
statement
INSERT INTO cl_src_intervals VALUES
(INTERVAL '1-2' YEAR TO MONTH, INTERVAL '1 02:03:04' DAY TO SECOND, 'a'),
(INTERVAL '0-0' YEAR TO MONTH, INTERVAL '0 00:00:00' DAY TO SECOND, 'a'),
(NULL, NULL, 'a')
query
SELECT grp, sort_array(collect_list(ym)), sort_array(collect_list(dt))
FROM cl_src_intervals GROUP BY grp ORDER BY grp
```
`timestamp_ntz` and ANSI interval literals both need Spark 3.4+, already the file's floor, so no `MinSparkVersion:` directive is required. (If the interval path can't be persisted to Parquet on some target
versions, gate that block with `-- MinSparkVersion: 3.4` explicitly.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3582715043
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1643,6 +1643,21 @@ trait CometBaseAggregate {
}
}
+ // CollectList/CollectSet declare their buffer as BinaryType in Spark but produce a native
Review Comment:
Isn't this block unreachable? The `missingCometProducer` guard at line 1632 already returns `None` for this case. A `PartialMerge` stage makes `consumesBuffers` true, and `CollectList`/`CollectSet` are always in `aggsNotSupportingMixedExecution` since they don't override `supportsMixedPartialFinal`. Looks like it predates the merge that brought that guard over from `main`. Can we drop it?
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -871,6 +871,24 @@ case class CometExecRule(session: SparkSession)
}
}
}
+
+ // CollectList/CollectSet round-trip an ArrayType buffer that Spark declares as BinaryType.
Review Comment:
Could you add a comment on why this is separate from the tagging block just above? They look like duplicates. The difference is that the block above only tags when the Final can't convert, and `canAggregateBeConverted` skips the child-native check, so an all-native `collect_list` Final converts and slips past it. This block is what catches that all-native distinct case, which is the Parquet half of the new suite test.
##########
spark/src/main/spark-4.2/org/apache/comet/shims/CometCollectShim.scala:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.shims
+
+import org.apache.spark.sql.catalyst.expressions.aggregate.{CollectList, CollectSet}
+
+/**
+ * Shim for the `ignoreNulls` flag on `CollectList` / `CollectSet`. Spark 4.2 added the field
+ * (`ignoreNulls = false` via `RESPECT NULLS`), so this shim reports the actual value. The serde
+ * falls back to Spark when it is `false`, since the native path always drops nulls.
+ */
+object CometCollectShim {
Review Comment:
These four files (3.4 through 4.1) are identical and only 4.2 differs. Could the 3.4/3.5 pair share the existing `spark-3.x` dir the way `CometTypeShim` does, so this is one fewer copy to keep in sync?
##########
spark/src/test/resources/sql-tests/expressions/aggregate/collect_list.sql:
##########
@@ -0,0 +1,405 @@
+-- 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.
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+-- collect_list result order is non-deterministic across partitions, so
+-- every query wraps the result in sort_array to make comparisons stable.
+
+-- ============================================================
+-- Setup: tables
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_int(i int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_int VALUES
+ (1, 'a'), (2, 'a'), (1, 'a'), (3, 'a'),
+ (4, 'b'), (4, 'b'), (NULL, 'b'), (5, 'b'),
+ (NULL, 'c'), (NULL, 'c')
+
+statement
+CREATE TABLE cl_src_nulls(val int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_nulls VALUES
Review Comment:
Could we add a `collect_list(x) FILTER (WHERE ...)` case, and something that exercises a map input? The map path hits the `Map` arm in `make_all_fields_nullable` that nothing covers yet. `sort_array` won't order a map, so it'd need `spark_answer_only` or a size check.
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -439,6 +440,22 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
}
}
+ /**
+ * Returns true if any aggregate function produces a native intermediate buffer whose Arrow type
+ * (e.g. ArrayType for CollectList/CollectSet) differs from the BinaryType that Spark declares
Review Comment:
The comment describes this as matching any agg with a native `ArrayType` buffer, but it only matches the collect functions. `Percentile` has the same buffer shape (see `adjustOutputForNativeState`) and isn't matched. It's fine in practice since `percentile` with distinct already passes, but the comment reads broader than the code. Worth narrowing the comment, or inlining the match since it's only used in one place now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#issuecomment-5007168233
> Thanks @andygrove I think we should leave `IGNORE NULLS` handling to Spark itself which would fail
We need to handle `RESPECT NULLS` in Spark 4.2 though. See comment https://github.com/apache/datafusion-comet/pull/4720#issuecomment-5005033208
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#issuecomment-5071918768
Thanks for the reviews @mbutrovich and @comphead. I filed follow on https://github.com/apache/datafusion-comet/issues/5024 to address some of the review feedback
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3554754608
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -740,6 +740,38 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
}
}
+object CometCollectList extends CometAggregateExpressionSerde[CollectList] {
+
+ override def convert(
+ aggExpr: AggregateExpression,
+ expr: CollectList,
+ inputs: Seq[Attribute],
+ binding: Boolean,
+ conf: SQLConf): Option[ExprOuterClass.AggExpr] = {
+ val child = expr.children.head
Review Comment:
Addressed in 9e680c42f. Added a `CometCollectShim.ignoreNulls` shim across the per-version source dirs: 3.4 through 4.1 have no such field so it returns `true`, and the spark-4.2 shim reads `agg.ignoreNulls`. `getSupportLevel` on both `CometCollectList` and `CometCollectSet` now returns `Unsupported` when `ignoreNulls = false`, so `collect_list(x) RESPECT NULLS` / `collect_set(x) RESPECT NULLS` on 4.2 falls back to Spark instead of silently dropping nulls.
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -413,6 +414,22 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
}
}
+ /**
+ * Returns true if any aggregate function produces a native intermediate buffer whose Arrow type
+ * (e.g. ArrayType for CollectList/CollectSet) differs from the BinaryType that Spark declares
+ * for its serialized TypedImperativeAggregate buffer. Comet cannot interpret Spark's Binary
+ * buffer for these functions, and cannot yet represent the buffer consistently across the
+ * intermediate PartialMerge stages of a multi-stage aggregate (issue #4724). Such aggregates
+ * are therefore only safe to run natively when every stage runs in Comet and there are at most
+ * two stages (Partial + Final).
+ */
+ def hasIncompatibleBufferAgg(aggExprs: Seq[AggregateExpression]): Boolean = {
Review Comment:
Renamed to `hasNativeArrayBufferAgg` in 9e680c42f.
##########
spark/src/test/resources/sql-tests/expressions/aggregate/collect_list.sql:
##########
@@ -0,0 +1,405 @@
+-- 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.
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+-- collect_list result order is non-deterministic across partitions, so
+-- every query wraps the result in sort_array to make comparisons stable.
+
+-- ============================================================
+-- Setup: tables
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_int(i int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_int VALUES
+ (1, 'a'), (2, 'a'), (1, 'a'), (3, 'a'),
+ (4, 'b'), (4, 'b'), (NULL, 'b'), (5, 'b'),
+ (NULL, 'c'), (NULL, 'c')
+
+statement
+CREATE TABLE cl_src_nulls(val int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_nulls VALUES
+ (NULL, 'a'), (NULL, 'a'), (NULL, 'b'), (1, 'b')
+
+statement
+CREATE TABLE cl_src_empty(val int) USING parquet
+
+statement
+CREATE TABLE cl_src_single(val int) USING parquet
+
+statement
+INSERT INTO cl_src_single VALUES (42)
+
+statement
+CREATE TABLE cl_src_dupes(val int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_dupes VALUES (7, 'a'), (7, 'a'), (7, 'a'), (8, 'b'), (9, 'b')
+
+-- ============================================================
+-- Basic: integer (global aggregate, no GROUP BY) — duplicates kept
+-- ============================================================
+
+query
+SELECT sort_array(collect_list(i)) FROM cl_src_int
+
+-- ============================================================
+-- GROUP BY: integer per group
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(i)) FROM cl_src_int GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- NULLs: nulls are dropped; all-NULL group returns empty array
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(val)) FROM cl_src_nulls GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Empty table: returns empty array
+-- ============================================================
+
+query
+SELECT sort_array(collect_list(val)) FROM cl_src_empty
+
+-- ============================================================
+-- Single value
+-- ============================================================
+
+query
+SELECT sort_array(collect_list(val)) FROM cl_src_single
+
+-- ============================================================
+-- All duplicates in a group — collect_list keeps repeats
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(val)) FROM cl_src_dupes GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Boolean (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_bool(v boolean, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_bool VALUES
+ (true, 'a'), (false, 'a'), (true, 'a'), (NULL, 'a'),
+ (NULL, 'b'), (true, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_bool GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Byte / Short (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_small(b tinyint, s smallint, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_small VALUES
+ (1, 100, 'a'), (2, 200, 'a'), (1, 100, 'a'), (NULL, NULL, 'a'),
+ (3, 300, 'b'), (NULL, 300, 'b')
+
+query
+SELECT grp, sort_array(collect_list(b)) FROM cl_src_small GROUP BY grp ORDER BY grp
+
+query
+SELECT grp, sort_array(collect_list(s)) FROM cl_src_small GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Int / BigInt (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_intbig(i int, bi bigint, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_intbig VALUES
+ (10, 1000000000000, 'a'), (20, 2000000000000, 'a'),
+ (10, 1000000000000, 'a'), (NULL, NULL, 'a'),
+ (30, 3000000000000, 'b'), (30, NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(i)) FROM cl_src_intbig GROUP BY grp ORDER BY grp
+
+query
+SELECT grp, sort_array(collect_list(bi)) FROM cl_src_intbig GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Float (with NULLs, NaN, Inf, -Inf, +0, -0)
+-- collect_list keeps duplicates verbatim, so floating-point is fine.
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_float(v float, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_float VALUES
+ (1.5, 'a'), (2.5, 'a'), (1.5, 'a'), (NULL, 'a'),
+ (CAST('NaN' AS FLOAT), 'b'), (CAST('NaN' AS FLOAT), 'b'), (1.0, 'b'),
+ (CAST('Infinity' AS FLOAT), 'c'), (CAST('-Infinity' AS FLOAT), 'c'),
+ (CAST('Infinity' AS FLOAT), 'c'),
+ (CAST(0.0 AS FLOAT), 'd'), (CAST(-0.0 AS FLOAT), 'd'), (1.0, 'd'), (NULL, 'd')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_float GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Double (with NULLs, NaN, Inf, -Inf, +0, -0)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_double(v double, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_double VALUES
+ (1.1, 'a'), (2.2, 'a'), (1.1, 'a'), (NULL, 'a'),
+ (CAST('NaN' AS DOUBLE), 'b'), (CAST('NaN' AS DOUBLE), 'b'), (1.0, 'b'),
+ (CAST('Infinity' AS DOUBLE), 'c'), (CAST('-Infinity' AS DOUBLE), 'c'),
+ (CAST('Infinity' AS DOUBLE), 'c'),
+ (0.0, 'd'), (-0.0, 'd'), (1.0, 'd'), (NULL, 'd')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_double GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- String (with NULLs and empty string)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_string(v string, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_string VALUES
+ ('hello', 'a'), ('world', 'a'), ('hello', 'a'), (NULL, 'a'),
+ ('', 'b'), ('x', 'b'), ('', 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_string GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Binary (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_binary(v binary, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_binary VALUES
+ (X'CAFE', 'a'), (X'BABE', 'a'), (X'CAFE', 'a'), (NULL, 'a'),
+ (X'', 'b'), (X'FF', 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_binary GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Decimal (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_decimal(v decimal(10,2), grp string) USING parquet
+
+statement
+INSERT INTO cl_src_decimal VALUES
+ (1.50, 'a'), (2.50, 'a'), (1.50, 'a'), (NULL, 'a'),
+ (0.00, 'b'), (99999999.99, 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_decimal GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Date (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_date(v date, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_date VALUES
+ (DATE '2024-01-01', 'a'), (DATE '2024-06-15', 'a'), (DATE '2024-01-01', 'a'), (NULL, 'a'),
+ (DATE '1970-01-01', 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_date GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Timestamp (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_ts(v timestamp, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_ts VALUES
+ (TIMESTAMP '2024-01-01 00:00:00', 'a'), (TIMESTAMP '2024-06-15 12:30:00', 'a'),
+ (TIMESTAMP '2024-01-01 00:00:00', 'a'), (NULL, 'a'),
+ (TIMESTAMP '1970-01-01 00:00:00', 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_ts GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Mixed with other aggregates
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(i)), count(*), sum(i)
+FROM cl_src_int GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Multiple collect_list in the same query
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_multi(a int, b string, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_multi VALUES
+ (1, 'x', 'g1'), (2, 'y', 'g1'), (1, 'x', 'g1'),
+ (3, 'z', 'g2'), (NULL, NULL, 'g2')
+
+query
+SELECT grp, sort_array(collect_list(a)), sort_array(collect_list(b))
+FROM cl_src_multi GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- DISTINCT: deduplicates before collecting (different planner path)
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(DISTINCT i)) FROM cl_src_int GROUP BY grp ORDER BY grp
Review Comment:
Agreed, no change needed. The lone `collect_list(DISTINCT i)` path stays fully native, and the aggregate SQL-file tests plus the `CometAggregateSuite` distinct-plus-collect fallback test still pass after merging the latest `apache/main` (which reworked the mixed-execution guard).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3554800962
##########
dev/diffs/3.5.8.diff:
##########
@@ -266,7 +266,21 @@ index 6f3090d8908..a0e9309888d 100644
}
// test case for ObjectHashAggregate and SortAggregate
-@@ -793,7 +796,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -771,7 +774,12 @@ class DataFrameAggregateSuite extends QueryTest
+ }
+
+ test("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") {
+- withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") {
++ withSQLConf(
++ SQLConf.USE_OBJECT_HASH_AGG.key -> "true",
++ // Comet runs collect_list/collect_set natively (CometHashAggregateExec), so the
++ // ObjectHashAggregateExec this test asserts on is no longer present. Disable Comet
++ // to preserve the original plan-shape assertions.
++ "spark.comet.enabled" -> "false") {
Review Comment:
So we're disabling Comet? That doesn't seem like the right approach.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3555139215
##########
dev/diffs/3.5.8.diff:
##########
@@ -266,7 +266,21 @@ index 6f3090d8908..a0e9309888d 100644
}
// test case for ObjectHashAggregate and SortAggregate
-@@ -793,7 +796,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -771,7 +774,12 @@ class DataFrameAggregateSuite extends QueryTest
+ }
+
+ test("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") {
+- withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") {
++ withSQLConf(
++ SQLConf.USE_OBJECT_HASH_AGG.key -> "true",
++ // Comet runs collect_list/collect_set natively (CometHashAggregateExec), so the
++ // ObjectHashAggregateExec this test asserts on is no longer present. Disable Comet
++ // to preserve the original plan-shape assertions.
++ "spark.comet.enabled" -> "false") {
Review Comment:
Fixed, but also found a new bug. Will keep in draft until that is resolved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#issuecomment-5005033208
This guard is for `RESPECT NULLS` (`ignoreNulls = false`), not `IGNORE NULLS`, and it only becomes reachable on Spark 4.2.
Spark 4.2 added an `ignoreNulls` field to `CollectList`/`CollectSet`. `FunctionResolution.applyIgnoreNulls` now has explicit cases for them:
```scala
case collectList: CollectList => collectList.copy(ignoreNulls = ignoreNulls)
case collectSet: CollectSet => collectSet.copy(ignoreNulls = ignoreNulls)
```
so `SELECT collect_list(c) RESPECT NULLS FROM ...` is accepted on 4.2 and produces `CollectList(ignoreNulls = false)`, whose buffer preserves nulls (`bufferContainsNull = !ignoreNulls`). Comet's native aggregate always drops nulls, so it has to fall back in that case, which is what this branch does.
Your `IGNORE NULLS` example throws because it runs on Spark 4.1 or earlier, where `CollectList` isn't in the `applyIgnoreNulls` match and instead hits `case _ if ignoreNulls => throw ... "IGNORE NULLS"`. On those versions the field doesn't exist, so `CometCollectShim` hardcodes `ignoreNulls = true` and this branch is an unreachable no-op. It's routed through the shim precisely so it only activates on 4.2. (On 4.2 your exact `IGNORE NULLS` query is actually accepted too, since it resolves to the default.)
I'll add a 4.2-only SQL test exercising `collect_list(c) RESPECT NULLS` plus a code comment spelling out the version split so this is clearer in the source.
For transparency: I used an LLM to help trace the Spark 4.2 resolution path while working through this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#issuecomment-5005107973
Pushed `a499164d0` adding test coverage and clearer comments for the `RESPECT NULLS` fallback.
New SQL file test `spark/src/test/resources/sql-tests/expressions/aggregate/collect_respect_nulls.sql`, gated with `-- MinSparkVersion: 4.2`:
- `collect_list(v) RESPECT NULLS` and `collect_set(v) RESPECT NULLS` (global, per-group, and an all-null group) use `query expect_fallback(...)`, asserting both the exact fallback reason (`... with RESPECT NULLS (ignoreNulls = false) is not supported`) and Spark-identical results.
- A closing default-behavior `collect_list` query stays native (`checkSparkAnswerAndOperator`) to prove the fallback is specific to `RESPECT NULLS` rather than a blanket collect fallback.
This runs in the `Spark 4.2, JDK 17` job's `expressions` bucket (which includes `CometSqlFileTestSuite`) and is skipped on 3.4 through 4.1, where the `ignoreNulls` field does not exist and the branch is an unreachable no-op.
I also expanded the `getSupportLevel` comments on `CometCollectList`/`CometCollectSet` to spell out that the fallback is only reachable on 4.2+.
For transparency: the analysis and these tests were done with LLM assistance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3604323347
##########
spark/src/main/spark-3.x/org/apache/comet/shims/CometCollectShim.scala:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.shims
+
+import org.apache.spark.sql.catalyst.expressions.aggregate.{CollectList, CollectSet}
+
+/**
+ * Shim for the `ignoreNulls` flag on `CollectList` / `CollectSet`. Spark 3.4 through 4.1 have no
+ * such field: these aggregates always drop null inputs, so this shim reports `true`. Spark 4.2
+ * added `ignoreNulls` (settable to `false` via `RESPECT NULLS`), handled by the spark-4.2 shim.
+ */
+object CometCollectShim {
+ def ignoreNulls(agg: CollectList): Boolean = true
Review Comment:
we shouldn't be having that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#issuecomment-5023178030
@comphead I responded to your comments - could you take another look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native collect_list / array_agg aggregate [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3506854700
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -740,6 +740,38 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
}
}
+object CometCollectList extends CometAggregateExpressionSerde[CollectList] {
+
+ override def convert(
+ aggExpr: AggregateExpression,
+ expr: CollectList,
+ inputs: Seq[Attribute],
+ binding: Boolean,
+ conf: SQLConf): Option[ExprOuterClass.AggExpr] = {
+ val child = expr.children.head
Review Comment:
This does not check `CollectList.ignoreNulls`, and the upstream `SparkCollectList` hardcodes `ignore_nulls = true` (`datafusion/spark/src/function/aggregate/collect.rs`). On Spark 3.4 through 4.1 `CollectList` has no `ignoreNulls` field so this is correct. On Spark 4.2 it gains `ignoreNulls: Boolean = true` and `RESPECT NULLS` becomes reachable from SQL. `FunctionResolution.applyIgnoreNulls` resolves it:
```scala
case collectList: CollectList => collectList.copy(ignoreNulls = ignoreNulls)
```
So `collect_list(x) RESPECT NULLS` sets `ignoreNulls = false`, which keeps null elements. Comet drops them and returns a different result from Spark with no fallback. 4.2 releases soon, so we might want to tackle this now. The field is absent on 3.4-4.1, so reading it needs a version shim (`CometExprShim`) rather than a direct `expr.ignoreNulls` reference. `CometCollectSet` has the same gap on 4.2, so consider covering both in the shim.
`CometFirst` (line 249) and `CometLast` (line 285) already thread `ignoreNulls` through the proto. `First`/`Last` carry the field on every supported Spark version, so no shim is needed there, which is the one extra wrinkle for `CollectList`/`CollectSet`. Same wiring shape applies once the field is shimmed.
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -413,6 +414,22 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
}
}
+ /**
+ * Returns true if any aggregate function produces a native intermediate buffer whose Arrow type
+ * (e.g. ArrayType for CollectList/CollectSet) differs from the BinaryType that Spark declares
+ * for its serialized TypedImperativeAggregate buffer. Comet cannot interpret Spark's Binary
+ * buffer for these functions, and cannot yet represent the buffer consistently across the
+ * intermediate PartialMerge stages of a multi-stage aggregate (issue #4724). Such aggregates
+ * are therefore only safe to run natively when every stage runs in Comet and there are at most
+ * two stages (Partial + Final).
+ */
+ def hasIncompatibleBufferAgg(aggExprs: Seq[AggregateExpression]): Boolean = {
Review Comment:
Nice to factor this out and share it between `CometExecRule` and `operators.scala` instead of duplicating. Minor naming thought: "incompatible buffer" is a bit generic. The method really means "produces a native ArrayType state where Spark declares BinaryType, so it cannot round-trip a Spark-produced intermediate buffer." Something like `hasNativeArrayBufferAgg` would carry more of that at the call sites. The doc comment is clear either way, so take it or leave it.
##########
spark/src/test/resources/sql-tests/expressions/aggregate/collect_list.sql:
##########
@@ -0,0 +1,405 @@
+-- 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.
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+-- collect_list result order is non-deterministic across partitions, so
+-- every query wraps the result in sort_array to make comparisons stable.
+
+-- ============================================================
+-- Setup: tables
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_int(i int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_int VALUES
+ (1, 'a'), (2, 'a'), (1, 'a'), (3, 'a'),
+ (4, 'b'), (4, 'b'), (NULL, 'b'), (5, 'b'),
+ (NULL, 'c'), (NULL, 'c')
+
+statement
+CREATE TABLE cl_src_nulls(val int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_nulls VALUES
+ (NULL, 'a'), (NULL, 'a'), (NULL, 'b'), (1, 'b')
+
+statement
+CREATE TABLE cl_src_empty(val int) USING parquet
+
+statement
+CREATE TABLE cl_src_single(val int) USING parquet
+
+statement
+INSERT INTO cl_src_single VALUES (42)
+
+statement
+CREATE TABLE cl_src_dupes(val int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_dupes VALUES (7, 'a'), (7, 'a'), (7, 'a'), (8, 'b'), (9, 'b')
+
+-- ============================================================
+-- Basic: integer (global aggregate, no GROUP BY) — duplicates kept
+-- ============================================================
+
+query
+SELECT sort_array(collect_list(i)) FROM cl_src_int
+
+-- ============================================================
+-- GROUP BY: integer per group
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(i)) FROM cl_src_int GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- NULLs: nulls are dropped; all-NULL group returns empty array
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(val)) FROM cl_src_nulls GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Empty table: returns empty array
+-- ============================================================
+
+query
+SELECT sort_array(collect_list(val)) FROM cl_src_empty
+
+-- ============================================================
+-- Single value
+-- ============================================================
+
+query
+SELECT sort_array(collect_list(val)) FROM cl_src_single
+
+-- ============================================================
+-- All duplicates in a group — collect_list keeps repeats
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(val)) FROM cl_src_dupes GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Boolean (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_bool(v boolean, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_bool VALUES
+ (true, 'a'), (false, 'a'), (true, 'a'), (NULL, 'a'),
+ (NULL, 'b'), (true, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_bool GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Byte / Short (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_small(b tinyint, s smallint, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_small VALUES
+ (1, 100, 'a'), (2, 200, 'a'), (1, 100, 'a'), (NULL, NULL, 'a'),
+ (3, 300, 'b'), (NULL, 300, 'b')
+
+query
+SELECT grp, sort_array(collect_list(b)) FROM cl_src_small GROUP BY grp ORDER BY grp
+
+query
+SELECT grp, sort_array(collect_list(s)) FROM cl_src_small GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Int / BigInt (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_intbig(i int, bi bigint, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_intbig VALUES
+ (10, 1000000000000, 'a'), (20, 2000000000000, 'a'),
+ (10, 1000000000000, 'a'), (NULL, NULL, 'a'),
+ (30, 3000000000000, 'b'), (30, NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(i)) FROM cl_src_intbig GROUP BY grp ORDER BY grp
+
+query
+SELECT grp, sort_array(collect_list(bi)) FROM cl_src_intbig GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Float (with NULLs, NaN, Inf, -Inf, +0, -0)
+-- collect_list keeps duplicates verbatim, so floating-point is fine.
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_float(v float, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_float VALUES
+ (1.5, 'a'), (2.5, 'a'), (1.5, 'a'), (NULL, 'a'),
+ (CAST('NaN' AS FLOAT), 'b'), (CAST('NaN' AS FLOAT), 'b'), (1.0, 'b'),
+ (CAST('Infinity' AS FLOAT), 'c'), (CAST('-Infinity' AS FLOAT), 'c'),
+ (CAST('Infinity' AS FLOAT), 'c'),
+ (CAST(0.0 AS FLOAT), 'd'), (CAST(-0.0 AS FLOAT), 'd'), (1.0, 'd'), (NULL, 'd')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_float GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Double (with NULLs, NaN, Inf, -Inf, +0, -0)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_double(v double, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_double VALUES
+ (1.1, 'a'), (2.2, 'a'), (1.1, 'a'), (NULL, 'a'),
+ (CAST('NaN' AS DOUBLE), 'b'), (CAST('NaN' AS DOUBLE), 'b'), (1.0, 'b'),
+ (CAST('Infinity' AS DOUBLE), 'c'), (CAST('-Infinity' AS DOUBLE), 'c'),
+ (CAST('Infinity' AS DOUBLE), 'c'),
+ (0.0, 'd'), (-0.0, 'd'), (1.0, 'd'), (NULL, 'd')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_double GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- String (with NULLs and empty string)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_string(v string, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_string VALUES
+ ('hello', 'a'), ('world', 'a'), ('hello', 'a'), (NULL, 'a'),
+ ('', 'b'), ('x', 'b'), ('', 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_string GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Binary (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_binary(v binary, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_binary VALUES
+ (X'CAFE', 'a'), (X'BABE', 'a'), (X'CAFE', 'a'), (NULL, 'a'),
+ (X'', 'b'), (X'FF', 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_binary GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Decimal (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_decimal(v decimal(10,2), grp string) USING parquet
+
+statement
+INSERT INTO cl_src_decimal VALUES
+ (1.50, 'a'), (2.50, 'a'), (1.50, 'a'), (NULL, 'a'),
+ (0.00, 'b'), (99999999.99, 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_decimal GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Date (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_date(v date, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_date VALUES
+ (DATE '2024-01-01', 'a'), (DATE '2024-06-15', 'a'), (DATE '2024-01-01', 'a'), (NULL, 'a'),
+ (DATE '1970-01-01', 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_date GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Timestamp (with NULLs)
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_ts(v timestamp, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_ts VALUES
+ (TIMESTAMP '2024-01-01 00:00:00', 'a'), (TIMESTAMP '2024-06-15 12:30:00', 'a'),
+ (TIMESTAMP '2024-01-01 00:00:00', 'a'), (NULL, 'a'),
+ (TIMESTAMP '1970-01-01 00:00:00', 'b'), (NULL, 'b')
+
+query
+SELECT grp, sort_array(collect_list(v)) FROM cl_src_ts GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Mixed with other aggregates
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(i)), count(*), sum(i)
+FROM cl_src_int GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Multiple collect_list in the same query
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_multi(a int, b string, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_multi VALUES
+ (1, 'x', 'g1'), (2, 'y', 'g1'), (1, 'x', 'g1'),
+ (3, 'z', 'g2'), (NULL, NULL, 'g2')
+
+query
+SELECT grp, sort_array(collect_list(a)), sort_array(collect_list(b))
+FROM cl_src_multi GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- DISTINCT: deduplicates before collecting (different planner path)
+-- ============================================================
+
+query
+SELECT grp, sort_array(collect_list(DISTINCT i)) FROM cl_src_int GROUP BY grp ORDER BY grp
Review Comment:
For the record, no change needed. A lone `collect_list(DISTINCT i)` does not create a `PartialMerge` stage on `collect_list` itself. Spark does the dedup with group-by-only stages and the collect still runs plain Partial then Final, so this query stays fully native and does not touch the `hasIncompatibleBufferAgg` fallback. The fallback path needs a distinct aggregate combined with the collect, which the new `CometAggregateSuite` test covers over both `LocalTableScan` and Parquet sources. Coverage across the two files is good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org