apache / apache/datafusion-comet
Struct-typed scalar subquery result takes the consuming projection off Comet (widened by Spark 4.2 MergeSubplans)
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
## Describe the bug
`CometScalarSubquery` declines any scalar subquery whose result type is a struct, because it
calls `supportedDataType` with the default `allowComplex = false`:
```scala
override def getSupportLevel(expr: ScalarSubquery): SupportLevel =
if (supportedDataType(expr.dataType)) {
Compatible()
} else {
Unsupported(Some(s"Unsupported data type: ${expr.dataType}"))
}
```
`spark/src/main/scala/org/apache/comet/serde/CometScalarSubquery.scala:33-38`
A struct-typed scalar subquery is not an exotic shape a user has to write: it is exactly what
`MergeScalarSubqueries` produces. That rule rewrites sibling one-row subplans into a single CTE
projecting `CreateNamedStruct(name1, attr1, name2, attr2, ...) AS mergedValue`, and rewrites each
original site to `GetStructField(ScalarSubquery(CTERelationRef), idx)`. So merging two subqueries
reliably takes the consuming projection off Comet.
This is long-standing and version-independent. What changed is the blast radius: **Spark 4.2
renames the rule to `MergeSubplans` and widens it to merge any one-row-returning subplan,
including a bare non-grouping `Aggregate` node** - not just `ScalarSubquery` plan trees. Queries
that contain no subqueries at all now get rewritten into this shape, and because the resulting
`Project` is not Comet, everything above it that needs a Comet child (a `Union`, and then any
aggregate whose intermediate buffer format is incompatible with Spark) goes off Comet too.
## Steps to reproduce
```scala
withParquetTable((0 until 100).map(i => (i, i * 2)), "tbl") {
sql("SELECT sum(s) FROM (" +
" SELECT max(_1) AS s FROM tbl UNION ALL" +
" SELECT min(_2) AS t FROM tbl)")
.queryExecution.executedPlan
}
```
Spark 4.1 - fully native, 11 of 11 eligible operators, 0 transitions:
```
CometHashAggregate
+- CometHashAggregate
+- CometUnion
:- CometHashAggregate
: +- CometExchange
: +- CometHashAggregate
: +- CometNativeScan parquet
+- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometNativeScan parquet
```
Spark 4.2 - 12 of 20 eligible operators, 2 transitions:
```
CometHashAggregate
+- CometColumnarExchange
+- HashAggregate
+- Union
:- Project [COMET: Unsupported data type: StructType(StructField(s,IntegerType,true),StructField(t,IntegerType,true))]
: : +- Subquery
: : +- CometProject
: : +- CometHashAggregate
: : +- CometExchange
: : +- CometHashAggregate
: : +- CometNativeScan parquet
: +- CometSparkRowToColumnar
: +- Scan OneRowRelation
+- Project [COMET: Unsupported data type: StructType(StructField(s,IntegerType,true),StructField(t,IntegerType,true))]
...
```
The two branches are aliased `s` and `t`, so the merged struct has distinct field names - the
duplicate-field-name limitation is not involved here. The sole fallback reason is the struct
result type.
For the version-independent half, this is enough on any supported version (4.1 and 4.2 produce
byte-identical fallback output for it):
```scala
sql("SELECT (SELECT max(_1) FROM tbl) AS a, (SELECT min(_1) FROM tbl) AS b")
// Project [COMET: Unsupported data type: StructType(StructField(max(_1),IntegerType,true),StructField(min(_1),IntegerType,true))]
```
## Expected behavior
A merged scalar subquery should not take the consuming projection off Comet. `GetStructField`
over a struct-typed `ScalarSubquery` is an ordinary struct field read; Comet supports `StructType`
elsewhere via `supportedDataType(dt, allowComplex = true)`.
## Additional context
Related, but each is a different gap - fixing any one of them alone does not fix this:
- #5586 - `named_struct` with duplicate field names falls back. Hit *in addition* to this issue
whenever the merged branches share an output alias (e.g. a `UNION ALL` where both sides say
`AS s`), which adds a second fallback reason on the CTE side. Fixing #5586 does not help the
distinct-alias case above.
- #4949 - Spark 4.2 TPC-DS q77a loses `CometUnion` and its aggregates. That issue attributes the
cascade to `Scan OneRowRelation` in the `Union` branches; the `OneRowRelation` is `MergeSubplans`
output (the rewritten site is `Project(GetStructField(ScalarSubquery(...)), OneRowRelation)`), so
the two are likely the same root cause. In the reproducer above the `OneRowRelation` is handled
fine via `CometSparkRowToColumnar` and the struct-typed subquery is what actually blocks.
- #5605 / #5783 - correctness bugs with duplicate-field-name structs in native shuffle and the
native Parquet scan. Worth checking before widening struct support anywhere.
Found while fixing CI on #4802, where this made a `hll_union_agg` test silently exercise Spark
instead of the native path on Spark 4.2 only.
Contributor guide
Assessment
This issue has not been assessed yet.