andygrove opened a new issue, #5605:
URL: https://github.com/apache/datafusion-comet/issues/5605
### Describe the bug
`nativeShuffleFailureReasons`'s `supportedSerializableDataType` accepts a
`StructType` whose fields have duplicate names, so native shuffle is chosen for
the plan. The batch is written and decoded natively without complaint, but the
JVM cannot import the decoded struct back across the Arrow C data interface,
and the task dies:
```
java.lang.IllegalStateException: ArrowArray struct has 2 children (expected
1)
at
org.apache.arrow.util.Preconditions.checkState(Preconditions.java:562)
at org.apache.arrow.c.ArrayImporter.doImport(ArrayImporter.java:92)
at org.apache.arrow.c.ArrayImporter.importArray(ArrayImporter.java:68)
at org.apache.arrow.c.ArrowImporter.importVector(ArrowImporter.java:56)
at org.apache.comet.vector.NativeUtil.importVector(NativeUtil.scala:257)
at org.apache.comet.vector.NativeUtil.getNextBatch(NativeUtil.scala:210)
at
org.apache.spark.sql.comet.execution.shuffle.NativeBatchDecoderIterator.fetchNext(NativeBatchDecoderIterator.scala:104)
at
org.apache.spark.sql.comet.execution.shuffle.NativeBatchDecoderIterator.hasNext(NativeBatchDecoderIterator.scala:80)
```
Java Arrow keys a struct vector's children by field name, so two same-named
children collapse into one and the import-time arity check fails.
The columnar shuffle path already excludes this shape, for exactly this
reason:
```scala
case StructType(fields) =>
fields.nonEmpty && fields.forall(f =>
supportedSerializableDataType(f.dataType)) &&
// Java Arrow stream reader cannot work on duplicate field name
fields.map(f => f.name).distinct.length == fields.length
```
`spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala:551-554`
The native path's copy of the predicate does not:
```scala
case StructType(fields) =>
fields.nonEmpty && fields.forall(f =>
supportedSerializableDataType(f.dataType))
```
`spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala:426-427`
`CometCreateNamedStruct` also guards against duplicate names
(`spark/src/main/scala/org/apache/comet/serde/structs.scala:42-43`, returning
`Unsupported`), so this is the one place among the three that knows about the
shape and does not exclude it. Because the expression serde declines, a plain
`SELECT named_struct('a', x, 'a', y) ...` projection falls back to Spark and
never reaches native shuffle. The gate is reached when the duplicate-named
struct arrives from something other than a Comet projection.
### Steps to reproduce
Reproduced on `199a910bd` (main), Spark 4.1 profile, macOS aarch64.
Two paths reach it.
**1. Cached relation feeding `CometSparkRowToColumnar`.** This needs no
non-default Comet configuration beyond shuffle being enabled, and reproduces
under the default `spark.comet.shuffle.mode=auto` as well as `native`:
```scala
val base = spark.range(50).selectExpr("id", "named_struct('a', id, 'a', id +
1) AS st")
base.cache()
base.repartition(4, col("id")).collect()
```
Chosen plan:
```
CometExchange hashpartitioning(id#2L, 4), REPARTITION_BY_NUM,
CometNativeShuffle
+- CometSparkRowToColumnar
+- InMemoryTableScan [id#2L, st#3]
+- InMemoryRelation [id#2L, st#3], StorageLevel(disk, memory,
deserialized, 1 replicas)
+- *(1) Project [id#2L, named_struct(a, id#2L, a, (id#2L +
1)) AS st#3]
+- *(1) Range (0, 50, step=1, splits=5)
```
`spark.comet.shuffle.convertFromSparkPlan.enabled` defaults to `true`, which
is what makes `CometSparkRowToColumnar` an eligible native-shuffle child here.
**2. `CometLocalTableScan`**
(`spark.comet.exec.localTableScan.enabled=true`). As a test in
`CometNativeShuffleSuite`:
```scala
test("native shuffle on struct data column with duplicate field names") {
withSQLConf(CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
val df = spark.sql(
"SELECT id, named_struct('a', id, 'a', id + 1) AS st " +
"FROM VALUES (1), (2), (3) AS t(id)")
checkShuffleAnswer(df.repartition(2, $"id"), 1)
}
}
```
Not reachable through Parquet: Spark rejects the schema at write time with
`COLUMN_ALREADY_EXISTS`.
### Expected behavior
`supportedSerializableDataType` on the native path should reject structs
with duplicate field names, the same way the columnar path does, so the shuffle
falls back to Spark and the query returns correct results instead of failing
the task. Spark itself permits duplicate field names in a struct, and the same
query succeeds with `spark.comet.shuffle.mode=jvm` and with Comet disabled.
### Additional context
Found while writing extra coverage around #5563 / #5564, which added the
first native-shuffle tests for struct data columns. The four tests added there
do not exercise this shape. I also probed the other uncovered native-shuffle
paths for struct columns (spill and merge, many small batches, `directRead`
both settings, range and round-robin partitioning, `map<int, struct>`,
`struct<null, int>`, binary/decimal/timestamp/date struct fields, three levels
of nesting, nested output from an aggregate, empty input) and they all behave
correctly. Duplicate field names were the only failure.
Prior art on the same Java Arrow limitation: #777 and #2457, both closed.
#1015 is the expression-side version.
Two notes for adjacent work:
- #5586 proposes routing `named_struct` with duplicate field names through
codegen dispatch instead of falling back, and says "duplicate names are not a
problem on the Arrow side ... Worth confirming with a test that the resulting
Arrow schema round-trips". This issue is evidence that they are a problem once
the value crosses the JVM Arrow FFI import boundary, so that round-trip test
matters, and enabling #5586 would widen the surface that reaches this gate.
- #5021 proposes consolidating the several data-type support predicates and
states "The predicates are correct today". This divergence is a counterexample
worth folding into that cleanup.
--
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]
