apache / apache/datafusion-comet

[Variant] Support Variant payloads through Comet shuffle and spill paths

Open
#5,434 1 comment 0 reactions 1 assignee Claimed by @peterxcli View on GitHub
area:shuffle enhancement
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.