unikdahal opened a new issue, #5413:
URL: https://github.com/apache/datafusion-comet/issues/5413
### 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 <col>`.
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.
--
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]
