apache / apache/datafusion-comet
Empty struct columns silently fall back to Spark instead of running natively
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
### Describe the bug
Six independent copies of the same guard reject a `StructType` with zero fields
(`fields.nonEmpty && fields.forall(...)`), collapsing "empty struct" into the same
`false` as a genuinely unsupported type. An empty struct is a legitimate,
trivially-serializable Arrow value -- zero child arrays, its own validity bitmap --
there's no technical reason to exclude it from the five Scala-side type-support checks
below. The sixth is different: it guards a real native panic, not just a planning-time
rejection (see below).
Found via Apache Iceberg's `_partition` metadata column, which is exactly `StructType()`
on an unpartitioned table. Any plan carrying that column (e.g. `_partition` selected
alongside `_file`/`_pos` for row-level MERGE/UPDATE/DELETE tracking) silently fell back
to Spark the moment it hit a shuffle, sink, or local-table-scan boundary -- no error,
just a quiet loss of native execution for everything downstream.
- `CometShuffleExchangeExec.scala` -- native shuffle's type check
- `CometShuffleExchangeExec.scala` -- columnar shuffle's type check (separate function,
same file)
- `CometSink.scala` -- the sink operator reading off a shuffle/exchange boundary, also
the gate for `LocalTableScanExec`
- `QueryPlanSerde.scala` -- the core expression-level type gate (hit via a
`Literal(_, StructType())` placeholder)
- `DataTypeSupport.scala` -- the general operator schema gate
- `serde/structs.scala` -- `from_json`'s target-schema check. Fixing this one alone
surfaced a real native panic: `native/spark-expr/src/json_funcs/from_json.rs` builds
the result via `StructArray::new(fields, arrays, nulls)`, which derives row count from
the first child array and panics when `fields` is empty (no child array to derive it
from). Fixed by branching to `StructArray::new_empty_fields(len, nulls)` when
`fields.is_empty()`.
### Steps to reproduce
```scala
val schema = StructType(
Seq(StructField("id", IntegerType), StructField("marker", StructType(Nil))))
val data = (0 until 50).map(i => Row(i, Row()))
val df = spark.createDataFrame(data.asJava, schema)
df.repartition(10, $"id").collect()
```
With `spark.comet.exec.shuffle.enabled=true`, the shuffle silently ran as plain Spark
`ShuffleExchangeExec`, not `CometShuffleExchangeExec` -- no fallback reason surfaced
unless `spark.comet.explain.fallback.log.enabled=true` was set, which then logged:
`unsupported shuffle data type StructType() for input `.
Separately, with the native `from_json` path opted in
(`spark.comet.expression.JsonToStructs.allowIncompatible=true`),
`from_json(col, 'struct<>')` crashed the executor with:
`CometNativeException: native panic: called Result::unwrap() on an Err value:
InvalidArgumentError("use StructArray::try_new_with_length or
StructArray::new_empty_fields...")`.
### Expected behavior
An empty-struct column should not, by itself, prevent native execution -- Comet should
run natively the same way it does for a non-empty struct, and never panic.
### Additional context
Discovered while getting native `MergeRowsExec` (Comet's row-level MERGE dispatch
operator) to engage for Iceberg merge-on-read tables: scan, join, and merge dispatch
all converted to native operators, but the chain silently broke the moment `_partition`
entered a shuffle boundary. Fix + tests (general-purpose, not Iceberg-specific) ready on
branch `fix-empty-struct-shuffle-support`, independent of any Iceberg-specific work.
Contributor guide
Research direction
Start with the six type-support checks named in CometShuffleExchangeExec.scala, CometSink.scala, QueryPlanSerde.scala, DataTypeSupport.scala, and serde/structs.scala, then reproduce the repartition example. Check the native from_json path in native/spark-expr/src/json_funcs/from_json.rs as well. Done means empty StructType columns remain on native execution paths and from_json no longer panics, with the relevant tests passing.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust, scala, spark
- Domain
- data-engineering, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 30/100