apache / apache/datafusion-comet
[Variant] Support Variant payloads through Comet shuffle and spill paths
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
### What is the problem the feature request solves?
Comet currently rejects every non-scan native operator whose schema contains `VariantType`, so Variant cannot be carried as a payload through Comet shuffle or its spill paths. Queries such as repartitioning by an ordinary `id` column while selecting `id, v` therefore fall back even though Spark permits Variant as a payload.
Spark rejects Variant when it is itself used as a partitioning expression ([analysis check](https://github.com/apache/spark/blob/v4.1.3/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala#L842-L856)); this issue must preserve that restriction. The gap is transport of a non-key Variant column. Comet's general [Variant operator gate](https://github.com/peterxcli/datafusion-comet/blob/c355fefd9c0b7e96d86523a7214bb2cdd47e1a55/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala#L734-L748) and the native/JVM [shuffle serializable-type checks](https://github.com/peterxcli/datafusion-comet/blob/c355fefd9c0b7e96d86523a7214bb2cdd47e1a55/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala#L407-L429) currently omit Variant.
### Describe the potential solution
- Admit direct, top-level Variant columns only as shuffle payloads; keep hash/range partitioning on Variant rejected like Spark.
- In native shuffle mode, preserve the parent `arrow.parquet.variant` Field marker, `[value, metadata]` child order, names, nullability, and metadata across IPC write/read and every spill/merge boundary.
- In JVM shuffle mode, decode Spark's dedicated UnsafeRow Variant payload into the canonical Arrow Variant Field rather than treating it as a generic Struct.
- Keep malformed or unsupported nested Variant schemas on explicit fallback.
Add Spark parity and plan tests for hash partitioning by a non-Variant key, round-robin, single-partition exchange, AQE/coalescing, and forced spill. Cover objects, arrays/scalars, Variant JSON null, SQL NULL, nullable parents, multiple Variant columns, and columns before/after Variant. Assert that partitioning directly by Variant still fails or falls back consistently with Spark.
### Additional context
#4295 explicitly calls for a separate Variant shuffle/spill issue. #5407 defines the whole-value scan representation; #5425 defines Variant-valued expression output.
C2R, Python transport, nested Variant, writes, and native Variant key semantics are separate concerns.
Contributor guide
Assessment
This issue has not been assessed yet.