Native shuffle accepts structs with duplicate field names, then fails importing the batch back to the JVM

[andygrove]

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:

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:

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:

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:

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:

• named_struct with duplicate field names falls back to Spark #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 named_struct with duplicate field names falls back to Spark #5586 would widen the surface that reaches this gate.
• Centralize data-type support predicates so new types don't have to be added in multiple places #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.