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

[andygrove]

Describe the bug

CometScalarSubquery declines any scalar subquery whose result type is a struct, because it
calls supportedDataType with the default allowComplex = false:

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

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):

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:

• named_struct with duplicate field names falls back to Spark #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 named_struct with duplicate field names falls back to Spark #5586 does not help the
  distinct-alias case above.
• Spark 4.2: OneRowRelation in Union branches forces Union and downstream aggregates off Comet (TPC-DS q77a) #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.
• Native shuffle accepts structs with duplicate field names, then fails importing the batch back to the JVM #5605 / Native Parquet scan multiplies rows for a struct with duplicate field names #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.