fix: support empty struct types

[unikdahal]

Which issue does this PR close?

Closes #5413.

Rationale for this change

Six copies of the same guard reject a zero-field StructType, treating it the same
as a genuinely unsupported type. It's a legitimate Arrow value. Iceberg's _partition
metadata column is exactly this shape on an unpartitioned table, so any plan carrying
it silently fell back to Spark at the first shuffle/sink/scan boundary.

What changes are included in this PR?

Removed the empty-struct exclusion from all six checks: native shuffle, columnar
shuffle, CometSink, QueryPlanSerde.supportedDataType, DataTypeSupport, and
from_json's target-schema check. The last one alone wasn't safe to fix Scala-side --
it uncovered a real native panic in from_json.rs (StructArray::new can't derive row
count with zero child arrays), fixed by branching to StructArray::new_empty_fields.

How are these changes tested?

Added new tests to test the fix.

[sunchao]

Summary

This PR treats zero-field structs as supported values across Comet's schema gates, adds native shuffle coverage for empty structs, and teaches native from_json to construct a top-level zero-field Arrow StructArray with an explicit row count.

Prior state and problem

Several independent support checks rejected StructType when it had no fields, causing otherwise representable schemas—such as Iceberg's empty _partition metadata struct for unpartitioned tables—to fall back at shuffle, sink, or serialization boundaries. Native from_json also used StructArray::new, whose length cannot be inferred when the result has no child arrays.

Design approach

The patch removes the non-empty requirement from the Scala support predicates and adds a top-level Rust branch that calls StructArray::new_empty_fields(num_rows, ...). The new shuffle tests exercise empty structs through native and columnar exchanges, while the JSON test covers a top-level struct<> target schema with and without dictionary encoding.

Correctness / compatibility analysis

The top-level empty-struct construction is consistent with Arrow 58.4.0 and preserves the input row count and validity buffer. The general support-check changes are also structurally consistent for empty structs. One native from_json gap remains, however: the recursive Scala predicate also enables nested empty structs, while the nested Rust builder still uses StructArray::new with zero child arrays and therefore panics.

Key design decisions

The patch correctly distinguishes an empty struct from an unsupported data type, preserves recursive validation for non-empty children, and limits the specialized Arrow constructor to the case where length cannot be inferred. The remaining decision is whether nested empty structs should be supported now or kept off the native from_json path until their builder uses the same explicit-length construction.

Implementation sketch

Six Scala guards are relaxed, the top-level JSON result chooses between new_empty_fields and StructArray::new, and focused shuffle/JSON tests are added. The nested FieldBuilder::Struct finalization path is unchanged.

Behavioral changes worth calling out

Plans carrying zero-field structs can remain in Comet through the updated boundaries instead of falling back. With native from_json opt-in enabled, top-level struct<> now works, but a schema such as struct<outer:struct<>> is also accepted and reaches an Arrow constructor that panics.

Suggested improvements

Handle zero-field nested structs in finish_builder using the nested validity-buffer length and add a regression test for a nested-empty schema. Alternatively, keep nested empty structs unsupported by the native JSON predicate until that construction path is implemented.

[sunchao]

[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]

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.

[sunchao]

[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]

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.

[sunchao]

[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]

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]

[P2] Restrict blank-input detection to JSON whitespace

With spark.comet.expression.JsonToStructs.allowIncompatible=true, trim() also treats non-JSON whitespace such as NBSP (U+00A0), vertical tab and form feed as a blank document. For a column containing only NBSP, the exact-head native expression now makes from_json(col, 'struct<>') IS NULL true, while Spark 3.5.2 and 4.0.4 return a non-null empty struct. This also regresses already-supported schemas such as a INT: both the base and previous native head return the non-null, all-null-fields struct, but this new branch returns SQL NULL. Restrict the blank check to JSON whitespace (space, tab, CR and LF) and add a non-JSON-whitespace regression row alongside the ordinary blank-input cases.

[unikdahal]

Fixed. Restricted the blank check to JSON whitespace (space/tab/CR/LF), verified against
Jackson's actual tokenizer behavior in current Spark source, not just the old SPARK-19543
comment. Added an NBSP regression row alongside the ordinary blank-input cases.

[sunchao]

Follow-up on 34083c79d1ed: two remaining empty-struct regressions.

[sunchao]

[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]

[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]

[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]

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]

[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]

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.

[sunchao]

[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]

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