Skip to content

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

Description

@unikdahal

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

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.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions