apache / apache/datafusion-comet

Empty struct columns silently fall back to Spark instead of running natively

Open
#5,413 0 comments 0 reactions 0 assignees View on GitHub
area:expressions area:shuffle enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.