apache / apache/datafusion-comet

Struct-typed scalar subquery result takes the consuming projection off Comet (widened by Spark 4.2 MergeSubplans)

Open
#5,834 1 comment 0 reactions 1 assignee Claimed by @LinSimon-901101 View on GitHub
area:expressions bug performance priority:medium requires-triage spark 4.2
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.