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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
