Skip to content

fix: support empty struct types #5414

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
unikdahal wants to merge 10 commits into
base: main
Choose a base branch
from
Open

fix: support empty struct types #5414

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
ce25a30
Support empty struct types
unikdahal Aug 22, 2026
5a5d4f3
Handle nested empty structs
unikdahal Aug 22, 2026
d763c74
Preserve NULL for blank input
unikdahal Aug 22, 2026
2fcee1c
Restrict blank check to JSON whitespace
unikdahal Aug 22, 2026
34083c7
Guard FIRST LAST on empty struct
unikdahal Aug 22, 2026
6775382
Guard collect_set and grouping keys
unikdahal Aug 22, 2026
1950465
Guard LAG LEAD empty struct default
unikdahal Aug 22, 2026
d168ea6
Fix literal empty struct crash
unikdahal Aug 22, 2026
332d9c2
Guard array_max min empty struct
unikdahal Aug 23, 2026
017266b
Guard RANGE frame empty struct order
unikdahal Aug 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 55 additions & 22 deletions native/spark-expr/src/json_funcs/from_json.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,17 @@ impl PhysicalExpr for FromJson {
}
}

/// Whether `s` is blank per Spark's own definition (SPARK-19543): Spark's `JacksonParser`
/// treats input as blank when the underlying Jackson `JsonParser` finds no first token, which
/// happens when only JSON whitespace precedes EOF. RFC 8259 defines JSON whitespace as exactly
/// space, tab, CR, and LF -- a narrower set than Rust's Unicode-aware `str::trim()` (which also
/// trims e.g. NBSP, U+2028) and than ASCII "control or space" (which also trims e.g. NUL, BEL,
/// vertical tab -- not JSON whitespace, so Jackson would fail to tokenize them and PERMISSIVE
/// mode would produce a non-null struct, not NULL).
fn is_blank(s: &str) -> bool {
s.trim_matches([' ', '\t', '\n', '\r']).is_empty()
}

/// Parse JSON string array into struct array
fn json_string_to_struct(arr: &Arc<dyn Array>, schema: &DataType) -> Result<ArrayRef> {
use arrow::array::StringArray;
Expand All @@ -150,27 +161,35 @@ fn json_string_to_struct(arr: &Arc<dyn Array>, schema: &DataType) -> Result<Arra
} else {
let json_str = string_array.value(row_idx);

// Parse JSON (PERMISSIVE mode: return null fields on error)
match serde_json::from_str::<serde_json::Value>(json_str) {
Ok(json_value) => {
if let serde_json::Value::Object(obj) = json_value {
// Struct is not null, extract each field
*struct_null = true;
for (field, builder) in fields.iter().zip(field_builders.iter_mut()) {
let field_value = obj.get(field.name());
append_field_value(builder, field, field_value)?;
if is_blank(json_str) {
// Blank input (empty or whitespace only) is NULL, not a struct with null
// fields (SPARK-19543) -- distinct from a non-blank string that fails to
// parse, which PERMISSIVE mode below turns into a struct with null fields.
*struct_null = false;
append_null_to_all_builders(&mut field_builders);
} else {
// Parse JSON (PERMISSIVE mode: return null fields on error)
match serde_json::from_str::<serde_json::Value>(json_str) {
Ok(json_value) => {
if let serde_json::Value::Object(obj) = json_value {
// Struct is not null, extract each field
*struct_null = true;
for (field, builder) in fields.iter().zip(field_builders.iter_mut()) {
let field_value = obj.get(field.name());
append_field_value(builder, field, field_value)?;
}
} else {
// Not an object -> struct with null fields
*struct_null = true;
append_null_to_all_builders(&mut field_builders);
}
} else {
// Not an object -> struct with null fields
}
Err(_) => {
// Parse error -> struct with null fields (PERMISSIVE mode)
*struct_null = true;
append_null_to_all_builders(&mut field_builders);
}
}
Err(_) => {
// Parse error -> struct with null fields (PERMISSIVE mode)
*struct_null = true;
append_null_to_all_builders(&mut field_builders);
}
}
}
}
Expand All @@ -180,11 +199,15 @@ fn json_string_to_struct(arr: &Arc<dyn Array>, schema: &DataType) -> Result<Arra
.map(finish_builder)
.collect::<Result<Vec<_>>>()?;
let null_buffer = NullBuffer::from(struct_nulls);
Ok(Arc::new(StructArray::new(
fields.clone(),
arrays,
Some(null_buffer),
)))
// `StructArray::new` derives its length from the first child array, so it panics when
// `fields` is empty (a legitimate zero-field target schema, e.g. `from_json(_, 'struct<>')`).
// `new_empty_fields` takes the length explicitly instead.
let struct_array: ArrayRef = if fields.is_empty() {
Arc::new(StructArray::new_empty_fields(num_rows, Some(null_buffer)))

@sunchao sunchao Aug 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Preserve Spark NULLs for blank JSON with an empty struct schema

With spark.comet.expression.JsonToStructs.allowIncompatible=true, this branch now makes from_json(col, 'struct<>') execute natively when col is '' or whitespace. The native parser marks every parse error as a valid struct, so the exact-head expression produces a non-null Row() and from_json(col, 'struct<>') IS NULL is false; Spark 3.5 and 4.0 intentionally return NULL for blank inputs (SPARK-19543). Before this change the empty schema took the Spark/codegen path, so this is a newly introduced wrong-result case. The added test uses only {}; please preserve the NULL validity bit for blank input and add empty/whitespace regression rows.

@unikdahal unikdahal Aug 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Blank/whitespace input now short-circuits to NULL before parsing, matching SPARK-19543, separate from non-blank malformed input (still PERMISSIVE null-fields). New regression rows cover blank, whitespace, non-blank-malformed, and SQL NULL, checked against real Spark output.

} else {
Arc::new(StructArray::new(fields.clone(), arrays, Some(null_buffer)))
};
Ok(struct_array)
}

/// Builder enum for different data types
Expand Down Expand Up @@ -393,7 +416,17 @@ fn finish_builder(builder: FieldBuilder) -> Result<ArrayRef> {
.map(finish_builder)
.collect::<Result<Vec<_>>>()?;
let null_buf = arrow::buffer::NullBuffer::from(null_buffer);
Arc::new(StructArray::new(fields, nested_arrays, Some(null_buf)))
// `StructArray::new` derives its row count from the first child array; a zero-field
// schema (e.g. a `struct<>`-typed field nested inside a larger schema) has no child
// arrays to derive it from, so the count is supplied explicitly here instead.
if fields.is_empty() {
Arc::new(StructArray::new_empty_fields(
null_buf.len(),
Some(null_buf),
))
} else {
Arc::new(StructArray::new(fields, nested_arrays, Some(null_buf)))
}
}
})
}
Expand Down
5 changes: 3 additions & 2 deletions spark/src/main/scala/org/apache/comet/DataTypeSupport.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,9 @@ trait DataTypeSupport {
CalendarIntervalType =>
true
case StructType(fields) =>
fields.nonEmpty && fields.forall(f =>
isTypeSupported(f.dataType, f.name, fallbackReasons))
// A struct's `fields` can be empty -- e.g. Iceberg's `_partition` metadata column is
// exactly that on an unpartitioned table. It's still a value Comet can represent.
fields.forall(f => isTypeSupported(f.dataType, f.name, fallbackReasons))

@sunchao sunchao Aug 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Guard empty structs in FIRST_VALUE/LAST_VALUE windows

This also admits empty-struct inputs to native windows that cannot handle them. Using the marker: struct<> LocalRelation from the new tests, enable spark.comet.exec.localTableScan.enabled=true with native shuffle and run SELECT first_value(marker) OVER () FROM t (no ORDER BY). The input can now stay native through CometWindowExec, which maps this to DataFusion 54.1's FirstValue. Its accumulator calls ScalarValue::compact(), whose compact_view_buffers struct branch reconstructs the array with StructArray::new even when there are no child fields, causing an Arrow panic. I reproduced this through the dependency's WindowExpr::evaluate on a valid three-row empty-struct batch; Spark returns three empty structs. LAST_VALUE fails identically, and the compaction is recursive, so nested/list/map-contained empty structs also fail. The old type gates kept these inputs on Spark. Please keep these windows on Spark for schemas containing empty structs until scalar compaction is fixed.

@unikdahal unikdahal Aug 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. CometFirst/CometLast now decline any schema containing an empty struct
(recursively through struct/array/map, matching where ScalarValue::compact would panic)
via a getSupportLevel check applies whether First/Last is used as a plain aggregate
or a window function, since both share the same serde. Added a regression test
reproducing your exact repro; confirmed no panic, correct fallback.

Also opened the actual fix upstream: apache/datafusion#24582.

@sunchao sunchao Aug 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Apply the empty-struct guard to collect_set too

Could we also reject empty-struct-containing inputs in CometCollectSet? With the marker: struct<> LocalRelation from the new tests, spark.comet.exec.localTableScan.enabled=true and native shuffle, SELECT collect_set(marker) FROM t is now admitted to native aggregation. SparkCollectSet wraps DataFusion 54.1's DistinctArrayAggAccumulator, which calls ScalarValue::compacted() for each non-null input and hits the same zero-field StructArray::new panic as FIRST/LAST. I reproduced this through the pinned accumulator with both top-level and nested empty structs, while Spark 3.5.2 and 4.0.4 return a one-element array. The ordinary Partial/Final plan has no PartialMerge stage, so the existing collect-buffer fallback does not contain it. Please reuse the recursive guard for collect_set and add a regression query.

@sunchao sunchao Aug 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Keep empty-struct grouping keys off native aggregation

With spark.comet.exec.localTableScan.enabled=true, SELECT DISTINCT marker FROM t for a LocalRelation containing nullable marker: struct<> now passes this gate and retains a native group-only/partial aggregate. Falling back the complex-key hash exchange does not revert that already-converted child. In pinned DataFusion 54.1, GroupValuesRows::emit calls dictionary_encode_if_necessary, whose struct branch uses StructArray::try_new with zero fields, so emitting the groups fails with Arrow's InvalidArgumentError. I reproduced this through the actual AggregateExec in Partial mode, both with no aggregate functions and with COUNT; Spark 3.5.2/4.0.4 return the expected empty-struct and NULL groups. Nested/list-contained empty-struct keys also fail. Please reject these grouping expressions or fix the group reconstruction before admitting them; the FIRST/LAST guard does not cover group-key emission.

@unikdahal unikdahal Aug 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. Rejects grouping keys containing an empty struct in both CometBaseAggregate.doConvert and the mirrored canAggregateBeConverted tag check (per its own WARNING comment). Added regression tests for GROUP BY and DISTINCT.

@sunchao sunchao Aug 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Guard nested-empty values in native scalar-ordering paths

With native local-table scan/shuffle enabled, let df contain id: INT and a: ARRAY<STRUCT<marker:STRUCT<>>>, with a = [Row(null), Row(Row())]. df.repartition(2, df("id")).selectExpr("array_max(a)") stays on Spark at the pinned base, but the exact-head planner converts it to CometProject -> CometNativeShuffle -> CometLocalTableScan. Spark 3.5.2/4.0.4 returns the element whose marker is {}; the pinned DataFusion 54.1 UDF instead returns the element whose marker is NULL.

ScalarValue::partial_cmp_struct recursively flattens structs, so the zero-field child contributes neither fields nor validity and those two values compare equal. The same comparator also misidentifies RANGE-window peers: with s: struct<e:struct<>> and a second integer order key, COUNT(*) OVER (ORDER BY s, k RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) counts {e:NULL} and {e:{}} together. The new compaction, grouping-reconstruction, and default-cast guards do not cover these ordering paths. Please keep the affected array extrema and RANGE order keys containing empty structs on Spark until native comparison preserves nested validity.

@unikdahal unikdahal Aug 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed both. Same root cause, two spots: ScalarValue::partial_cmp_struct flattens a struct to its leaf columns to compare, so a zero-field struct loses its own validity bit and a NULL element ties with a non-null {} element. Guarded array_max/array_min on element type, and RANGE frame ordering on the ORDER BY key type (the existing offset checks only covered explicit-offset bounds UNBOUNDED/CURRENT ROW skipped them entirely). Also checked sort_array and plain ORDER BY both use arrow's row-format comparator instead of ScalarValue::partial_cmp, which encodes struct-level validity independent of field count, so they're not affected by this one

case ArrayType(elementType, _) =>
isTypeSupported(elementType, ARRAY_ELEMENT, fallbackReasons)
case MapType(keyType, valueType, _) =>
Expand Down
3 changes: 2 additions & 1 deletion spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1071,7 +1071,8 @@ case class CometExecRule(session: SparkSession)
if (groupingExpressions.isEmpty && aggregateExpressions.isEmpty) return false

if (groupingExpressions.exists(e =>
SupportLevel.containsType(e.dataType, classOf[MapType]))) {
SupportLevel.containsType(e.dataType, classOf[MapType]) ||
SupportLevel.containsEmptyStruct(e.dataType))) {
return false
}

Expand Down
4 changes: 3 additions & 1 deletion spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -539,7 +539,9 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
case dt if isTimeType(dt) =>
true
case s: StructType if allowComplex =>
s.fields.nonEmpty && s.fields.map(_.dataType).forall(supportedDataType(_, allowComplex))
// A struct's `fields` can be empty -- e.g. Iceberg's `_partition` metadata column is
// exactly that on an unpartitioned table. It's still a value Comet can represent.
s.fields.map(_.dataType).forall(supportedDataType(_, allowComplex))

@sunchao sunchao Aug 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Preserve fallback for nested-array empty-struct literals

Could we keep literal fallback until these element types can be serialized? Over a Parquet-backed table, SELECT id, array(array(struct())) FROM t folds to a non-null array<array<struct<>>> literal. The widened predicate changes CometLiteral.getSupportLevel from Unsupported to Compatible, but makeListLiteral recursively reaches StructType() without a matching branch and throws scala.MatchError during planning. I reproduced that base/head difference using the complete literal serializer, with Spark 4.0.4 successfully executing the query over an id: bigint Parquet scan. The exception is not caught by the expression or operator conversion path. Please either implement this serialization or recursively restrict the literal element types, and add a regression test that keeps constant folding enabled.

@sunchao sunchao Aug 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Guard typed NULL array defaults in LEAD/LAG

This also admits CAST(NULL AS ARRAY<STRUCT<>>) as a window default. With native local-table scan/window execution enabled and an arr: array<struct<>> input, lag(arr, 1, CAST(NULL AS ARRAY<STRUCT<>>)) OVER (ORDER BY id) (and lead) passes the literal-default checks, but DataFusion 54.1 casts the typed NULL list to the input type. That recursively casts its zero-length struct child and errors with Cannot cast struct with 0 fields to 0 fields because there is no field name overlap, even when source and target datatypes are identical. The new FIRST/LAST guard does not run for these builtin windows. Spark 3.5.2/4.0.4 preserve the typed NULL and return the expected rows; I reproduced the failure with the pinned create_window_expr, while omitted/plain NULL defaults and nonempty-struct controls succeed. Please keep these defaults on Spark or fix their native coercion. This NULL path never invokes makeListLiteral, so fixing the already-reported non-null nested-array literal does not address it.

@unikdahal unikdahal Aug 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. Declines LAG/LEAD when the default expression's own type carries an empty struct - keyed on that, not the input type, so omitted/plain-NULL defaults (which don't hit the cast) stay native. Added regression tests for both the failing and the still-native forms.

case a: ArrayType if allowComplex =>
supportedDataType(a.elementType, allowComplex)
case m: MapType if allowComplex =>
Expand Down
19 changes: 19 additions & 0 deletions spark/src/main/scala/org/apache/comet/serde/SupportLevel.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,25 @@ object SupportLevel {
}
}

/**
* Whether `dt` is, or contains (recursively, through struct/array/map), a zero-field struct.
* Several DataFusion 54.1 code paths that reconstruct a struct -- `ScalarValue::compact`
* (backing the `FirstValue`/`LastValue`/`DistinctArrayAggAccumulator` accumulators),
* `GroupValuesRows::emit`'s dictionary-encoding step, and the cast applied to a window
* function's typed default value -- assume at least one child array (or a non-empty field-name
* overlap) to work from, and panic or error for a zero-field struct. Callers that would
* otherwise reach one of those paths decline such schemas instead; see
* https://github.com/apache/datafusion-comet/pull/5414.
*/
def containsEmptyStruct(dt: DataType): Boolean = dt match {
case StructType(fields) if fields.isEmpty => true
case StructType(fields) => fields.exists(f => containsEmptyStruct(f.dataType))
case ArrayType(elementType, _) => containsEmptyStruct(elementType)
case MapType(keyType, valueType, _) =>
containsEmptyStruct(keyType) || containsEmptyStruct(valueType)
case _ => false
}

/**
* Gate for [[CometConf.COMET_EXEC_STRICT_FLOATING_POINT]]: returns the standard incompatibility
* reason when strict mode is enabled and `dt` contains a float or double (at any nesting
Expand Down
38 changes: 32 additions & 6 deletions spark/src/main/scala/org/apache/comet/serde/aggregates.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,6 +240,9 @@ object CometFirst extends CometAggregateExpressionSerde[First] {
override def getCompatibleNotes(): Seq[String] = Seq(
"This function is not deterministic. Results may not match Spark.")

override def getSupportLevel(expr: First): SupportLevel =
AggSerde.firstLastSupportLevel(expr.dataType)

override def convert(
aggExpr: AggregateExpression,
first: First,
Expand Down Expand Up @@ -275,6 +278,9 @@ object CometLast extends CometAggregateExpressionSerde[Last] {
override def getCompatibleNotes(): Seq[String] = Seq(
"This function is not deterministic. Results may not match Spark.")

override def getSupportLevel(expr: Last): SupportLevel =
AggSerde.firstLastSupportLevel(expr.dataType)

override def convert(
aggExpr: AggregateExpression,
last: Last,
Expand Down Expand Up @@ -837,12 +843,20 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
" `spark.comet.expression.CollectSet.allowIncompatible=true` is set.")

override def getSupportLevel(expr: CollectSet): SupportLevel = {
// The native path always drops null inputs. Spark 4.2 added an `ignoreNulls` field to
// CollectSet that `RESPECT NULLS` sets to false, preserving nulls in the result; Comet
// cannot match that, so fall back. This branch is only reachable on Spark 4.2+: on 3.4
// through 4.1 the field does not exist, `RESPECT NULLS`/`IGNORE NULLS` are rejected at
// analysis time, and CometCollectShim.ignoreNulls hardcodes true, making this a no-op.
if (!CometCollectShim.ignoreNulls(expr)) {
// DataFusion's DistinctArrayAggAccumulator (backing SparkCollectSet) calls
// ScalarValue::compacted() per non-null input, hitting the same zero-field
// StructArray::new panic as First/Last -- see SupportLevel.containsEmptyStruct.
if (SupportLevel.containsEmptyStruct(expr.dataType)) {
Unsupported(
Some(
"collect_set on a schema containing an empty struct is not supported " +
"(DataFusion's ScalarValue::compacted panics on zero-field structs)"))
// The native path always drops null inputs. Spark 4.2 added an `ignoreNulls` field to
// CollectSet that `RESPECT NULLS` sets to false, preserving nulls in the result; Comet
// cannot match that, so fall back. This branch is only reachable on Spark 4.2+: on 3.4
// through 4.1 the field does not exist, `RESPECT NULLS`/`IGNORE NULLS` are rejected at
// analysis time, and CometCollectShim.ignoreNulls hardcodes true, making this a no-op.
} else if (!CometCollectShim.ignoreNulls(expr)) {
Unsupported(Some("collect_set with RESPECT NULLS (ignoreNulls = false) is not supported"))
} else {
SupportLevel
Expand Down Expand Up @@ -1030,6 +1044,18 @@ object AggSerde {
}
}

/** Shared support level for `First` / `Last`, which can't accept an empty struct. */
def firstLastSupportLevel(dt: DataType): SupportLevel = {
if (SupportLevel.containsEmptyStruct(dt)) {
Unsupported(
Some(
"FIRST/LAST on a schema containing an empty struct is not supported " +
"(DataFusion's ScalarValue::compact panics on zero-field structs)"))
} else {
Compatible()
}
}

/** Shared support level for `Min` / `Max` based on the result data type. */
def minMaxSupportLevel(dt: DataType): SupportLevel = {
if (!minMaxDataTypeSupported(dt)) {
Expand Down
20 changes: 20 additions & 0 deletions spark/src/main/scala/org/apache/comet/serde/arrays.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,17 @@ object CometArrayIntersect
}

object CometArrayMax extends CometExpressionSerde[ArrayMax] {
override def getSupportLevel(expr: ArrayMax): SupportLevel = {
// DataFusion's `ScalarValue::partial_cmp_struct` flattens a struct into its leaf columns to
// find the extreme element; a zero-field struct contributes no columns, so a NULL element
// and a non-null empty-struct element compare equal instead of ordering NULL out.
if (SupportLevel.containsEmptyStruct(expr.dataType)) {
Unsupported(Some("array_max on a schema containing an empty struct is not supported"))
} else {
Compatible()
}
}

override def convert(
expr: ArrayMax,
inputs: Seq[Attribute],
Expand All @@ -229,6 +240,15 @@ object CometArrayMax extends CometExpressionSerde[ArrayMax] {
}

object CometArrayMin extends CometExpressionSerde[ArrayMin] {
override def getSupportLevel(expr: ArrayMin): SupportLevel = {
// Same DataFusion comparator issue as CometArrayMax above.
if (SupportLevel.containsEmptyStruct(expr.dataType)) {
Unsupported(Some("array_min on a schema containing an empty struct is not supported"))
} else {
Compatible()
}
}

override def convert(
expr: ArrayMin,
inputs: Seq[Attribute],
Expand Down
25 changes: 17 additions & 8 deletions spark/src/main/scala/org/apache/comet/serde/literals.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,13 +24,12 @@ import java.lang
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal}
import org.apache.spark.sql.catalyst.util.ArrayData
import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, DateType, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampNTZType, TimestampType}
import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, DataType, DateType, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, MapType, NullType, ShortType, StringType, StructType, TimestampNTZType, TimestampType}
import org.apache.spark.unsafe.types.UTF8String

import com.google.protobuf.ByteString

import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
import org.apache.comet.DataTypeSupport.isComplexType
import org.apache.comet.serde.{CometExpressionSerde, Compatible, ExprOuterClass, LiteralOuterClass, SupportLevel, Unsupported}
import org.apache.comet.serde.QueryPlanSerde.{isTimeType, serializeDataType, supportedDataType}
import org.apache.comet.serde.Types.ListLiteral
Expand All @@ -40,6 +39,20 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
override def getUnsupportedReasons(): Seq[String] = Seq(
"Not all data types are supported for literal values")

/**
* Whether `makeListLiteral` can serialize an array literal whose (possibly nested) element type
* is `dt`. It has an explicit case per primitive type and recurses through `ArrayType`, but no
* case for `StructType` or `MapType` at all -- passing one through throws a `MatchError` at
* plan time. Checking this recursively (rather than only one level of nesting, as the caller
* below used to) keeps e.g. `array<array<struct<...>>>` -- empty struct or not -- off the
* literal path until `makeListLiteral` can actually encode it.
*/
private def isListLiteralElementSupported(dt: DataType): Boolean = dt match {
case a: ArrayType => isListLiteralElementSupported(a.elementType)
case _: StructType | _: MapType => false
case _ => true
}

override def getSupportLevel(expr: Literal): SupportLevel = {

if (supportedDataType(
Expand All @@ -48,12 +61,8 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {

// Nested literal support for native reader
// can be tracked https://github.com/apache/datafusion-comet/issues/1937
(expr.dataType
.isInstanceOf[ArrayType] && (!isComplexType(
expr.dataType.asInstanceOf[ArrayType].elementType) || expr.dataType
.asInstanceOf[ArrayType]
.elementType
.isInstanceOf[ArrayType])))) {
(expr.dataType.isInstanceOf[ArrayType] &&
isListLiteralElementSupported(expr.dataType.asInstanceOf[ArrayType].elementType)))) {
Compatible(None)
} else {
expr.dataType match {
Expand Down
4 changes: 3 additions & 1 deletion spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,9 @@ abstract class CometSink[T <: SparkPlan] extends CometOperatorSerde[T] {
protected final def supportedSinkDataType(dt: DataType): Boolean = dt match {
case _: YearMonthIntervalType | _: DayTimeIntervalType => true
case StructType(fields) =>
fields.nonEmpty && fields.forall(f => supportedSinkDataType(f.dataType))
// A struct's `fields` can be empty -- e.g. Iceberg's `_partition` metadata column is
// exactly that on an unpartitioned table. It's still a value the sink can serialize.
fields.forall(f => supportedSinkDataType(f.dataType))
case ArrayType(elementType, _) => supportedSinkDataType(elementType)
case MapType(keyType, valueType, _) =>
supportedSinkDataType(keyType) && supportedSinkDataType(valueType)
Expand Down
4 changes: 3 additions & 1 deletion spark/src/main/scala/org/apache/comet/serde/structs.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -250,7 +250,9 @@ object CometJsonToStructs extends CometCodegenDispatch[JsonToStructs] with Nativ

private def isSupportedSchema(dt: DataType): Boolean = dt match {
case StructType(fields) =>
fields.nonEmpty && fields.forall(f => isSupportedSchema(f.dataType))
// A struct's `fields` can be empty -- e.g. `from_json(col, 'struct<>')`'s target schema.
// With no fields to check, this holds vacuously.
fields.forall(f => isSupportedSchema(f.dataType))

@sunchao sunchao Aug 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Keep nested empty structs off the native path until they are constructed safely

Because this predicate is recursive, it now admits struct<outer:struct<>>, not just the top-level struct<> covered by the new test. The top-level Rust branch does not cover that shape: finish_builder still calls StructArray::new for every nested FieldBuilder::Struct; for an empty nested struct, builders yields zero child arrays. Arrow 58.4.0's StructArray::new unwraps try_new, which rejects no child arrays, so the query panics once this PR routes it native. Please add the same empty-fields construction in finish_builder (using its null-buffer length) and a nested-empty regression test, or keep nested empties unsupported here.

@unikdahal unikdahal Aug 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the quick, thorough review @sunchao .

Good catch, the recursive Scala predicate let struct<outer:struct<>> through while
finish_builder's nested struct branch still panicked on it. Fixed with the same
is_empty() check, using null_buf.len() for the row count. Added a regression test
covering the nested case. All 10 tests in CometJsonExpressionSuite pass.

case DataTypes.IntegerType | DataTypes.LongType | DataTypes.FloatType | DataTypes.DoubleType |
DataTypes.BooleanType | DataTypes.StringType =>
true
Expand Down
Loading