diff --git a/native/spark-expr/src/json_funcs/from_json.rs b/native/spark-expr/src/json_funcs/from_json.rs
index eaca6db0160..59ca3ae4dfe 100644
--- a/native/spark-expr/src/json_funcs/from_json.rs
+++ b/native/spark-expr/src/json_funcs/from_json.rs
@@ -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;
@@ -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);
-                }
             }
         }
     }
@@ -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)))
+    } else {
+        Arc::new(StructArray::new(fields.clone(), arrays, Some(null_buffer)))
+    };
+    Ok(struct_array)
 }
 
 /// Builder enum for different data types
@@ -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)))
+            }
         }
     })
 }
diff --git a/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala b/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala
index 019b4d629b9..68b877f38d1 100644
--- a/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala
+++ b/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala
@@ -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))
       case ArrayType(elementType, _) =>
         isTypeSupported(elementType, ARRAY_ELEMENT, fallbackReasons)
       case MapType(keyType, valueType, _) =>
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
index 3c2668cfe92..1ee163931f4 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
@@ -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
     }
 
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index 6802dfaa646..f293a12d88e 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -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))
     case a: ArrayType if allowComplex =>
       supportedDataType(a.elementType, allowComplex)
     case m: MapType if allowComplex =>
diff --git a/spark/src/main/scala/org/apache/comet/serde/SupportLevel.scala b/spark/src/main/scala/org/apache/comet/serde/SupportLevel.scala
index 60f2ff79345..4fe5b41a244 100644
--- a/spark/src/main/scala/org/apache/comet/serde/SupportLevel.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/SupportLevel.scala
@@ -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
diff --git a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala
index db435e5b5d5..23ad5457ed8 100644
--- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala
@@ -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,
@@ -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,
@@ -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
@@ -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)) {
diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
index 748b1cee231..5970271b4bd 100644
--- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
@@ -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],
@@ -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],
diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
index bac3631babe..d10673e6b8c 100644
--- a/spark/src/main/scala/org/apache/comet/serde/literals.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/literals.scala
@@ -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
@@ -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(
@@ -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 {
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
index f2026013cbf..968e7ff1d63 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
@@ -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)
diff --git a/spark/src/main/scala/org/apache/comet/serde/structs.scala b/spark/src/main/scala/org/apache/comet/serde/structs.scala
index 2f9619d491f..2ad257c52a3 100644
--- a/spark/src/main/scala/org/apache/comet/serde/structs.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/structs.scala
@@ -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))
     case DataTypes.IntegerType | DataTypes.LongType | DataTypes.FloatType | DataTypes.DoubleType |
         DataTypes.BooleanType | DataTypes.StringType =>
       true
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala
index 4d4abf3d2e0..dd155e34f13 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala
@@ -34,7 +34,7 @@ import com.google.common.base.Objects
 
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
-import org.apache.comet.serde.{AggSerde, CometOperatorSerde, LiteralOuterClass, OperatorOuterClass}
+import org.apache.comet.serde.{AggSerde, CometOperatorSerde, LiteralOuterClass, OperatorOuterClass, SupportLevel}
 import org.apache.comet.serde.OperatorOuterClass.Operator
 import org.apache.comet.serde.QueryPlanSerde.{aggExprToProto, exprToProto, scalarFunctionExprToProto, serializeDataType}
 
@@ -261,6 +261,16 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] {
           // https://github.com/apache/datafusion-comet/issues/4268
           withFallbackReason(windowExpr, "Lag default value must be a literal")
           (None, None, false)
+        case lag: Lag if SupportLevel.containsEmptyStruct(lag.default.dataType) =>
+          // An omitted or plain-NULL default is untyped (`NullType`) and unaffected. Only an
+          // explicitly typed default -- e.g. `CAST(NULL AS ARRAY<STRUCT<>>)` -- carries the
+          // empty struct into DataFusion's cast of the default to the input type, which errors
+          // on a zero-field struct even when source and target agree ("no field name overlap")
+          // -- see SupportLevel.containsEmptyStruct.
+          withFallbackReason(
+            windowExpr,
+            "LAG with an empty-struct-typed default value is not supported")
+          (None, None, false)
         case lag: Lag =>
           val inputExpr = exprToProto(lag.input, output)
           val offsetExpr = exprToProto(lag.inputOffset, output)
@@ -271,6 +281,12 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] {
           // https://github.com/apache/datafusion-comet/issues/4268
           withFallbackReason(windowExpr, "Lead default value must be a literal")
           (None, None, false)
+        case lead: Lead if SupportLevel.containsEmptyStruct(lead.default.dataType) =>
+          // Same DataFusion cast issue as the LAG case above.
+          withFallbackReason(
+            windowExpr,
+            "LEAD with an empty-struct-typed default value is not supported")
+          (None, None, false)
         case lead: Lead =>
           val inputExpr = exprToProto(lead.input, output)
           val offsetExpr = exprToProto(lead.offset, output)
@@ -353,6 +369,22 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] {
       }
     }
 
+    // DataFusion's `ScalarValue::partial_cmp_struct` flattens a struct into its leaf columns to
+    // compare RANGE peers; a zero-field struct contributes no columns, so a NULL struct and a
+    // non-null empty struct compare equal instead of ordering NULL first. That misgroups RANGE
+    // peers for any RANGE frame on such an ORDER BY key, not just the explicit-offset ones
+    // checked below (whose bounds are UNBOUNDED/CURRENT ROW and so skip that check entirely).
+    f match {
+      case SpecifiedWindowFrame(RangeFrame, _, _)
+          if windowExpr.windowSpec.orderSpec.exists(o =>
+            SupportLevel.containsEmptyStruct(o.dataType)) =>
+        withFallbackReason(
+          windowExpr,
+          "RANGE frame ordering on a schema containing an empty struct is not supported")
+        return None
+      case _ =>
+    }
+
     // Comet's native window planner ships RANGE frame offsets as
     // ScalarValue::Int64, but a couple of ORDER BY types don't tolerate that:
     //   - DATE: arrow-arith requires an Interval RHS for Date32 arithmetic,
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
index 84f313ad37c..09add7b7949 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
@@ -419,7 +419,10 @@ object CometShuffleExchangeExec
       case dt if isTimeType(dt) =>
         true
       case StructType(fields) =>
-        fields.nonEmpty && fields.forall(f => supportedSerializableDataType(f.dataType))
+        // A struct's `fields` can be empty -- e.g. Iceberg's `_partition` metadata column is
+        // exactly that on an unpartitioned table. Arrow represents it as zero child arrays plus
+        // its own validity bitmap, which native shuffle serializes like any other struct.
+        fields.forall(f => supportedSerializableDataType(f.dataType))
       case ArrayType(elementType, _) =>
         supportedSerializableDataType(elementType)
       case MapType(keyType, valueType, _) =>
@@ -544,10 +547,12 @@ object CometShuffleExchangeExec
       case dt if isTimeType(dt) =>
         true
       case StructType(fields) =>
-        fields.nonEmpty && fields.forall(f => supportedSerializableDataType(f.dataType)) &&
+        // A struct's `fields` can be empty -- e.g. Iceberg's `_partition` metadata column is
+        // exactly that on an unpartitioned table. The distinct-name check below holds
+        // vacuously when there are no fields to compare.
+        fields.forall(f => supportedSerializableDataType(f.dataType)) &&
         // Java Arrow stream reader cannot work on duplicate field name
-        fields.map(f => f.name).distinct.length == fields.length &&
-        fields.nonEmpty
+        fields.map(f => f.name).distinct.length == fields.length
       case ArrayType(elementType, _) =>
         supportedSerializableDataType(elementType)
       case MapType(keyType, valueType, _) =>
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
index d91ad02a6cb..a8990d8780c 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
@@ -1689,6 +1689,16 @@ trait CometBaseAggregate {
       return None
     }
 
+    if (groupingExpressions.exists(expr => SupportLevel.containsEmptyStruct(expr.dataType))) {
+      // DataFusion's `GroupValuesRows::emit` dictionary-encodes struct-typed group keys via
+      // `StructArray::try_new`, which errors for a zero-field struct -- see
+      // SupportLevel.containsEmptyStruct.
+      withFallbackReason(
+        aggregate,
+        "Grouping on a schema containing an empty struct is not supported")
+      return None
+    }
+
     if (groupingExpressions.exists(expr => isStringCollationType(expr.dataType))) {
       // Collation-aware grouping requires collation-aware hashing/equality; Comet only
       // compares raw bytes, which would put rows that compare equal under the collation
diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
index 05a6e8e650d..e75314f5ceb 100644
--- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
@@ -19,16 +19,17 @@
 
 package org.apache.comet
 
+import scala.jdk.CollectionConverters._
 import scala.util.Random
 
 import org.apache.hadoop.fs.Path
-import org.apache.spark.sql.CometTestBase
+import org.apache.spark.sql.{CometTestBase, Row}
 import org.apache.spark.sql.catalyst.expressions.{ArrayAppend, ArrayExcept, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayRepeat}
 import org.apache.spark.sql.catalyst.expressions.{ArrayContains, ArrayRemove}
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.ArrayType
+import org.apache.spark.sql.types.{ArrayType, IntegerType, StructField, StructType}
 
 import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus}
 import org.apache.comet.DataTypeSupport.isComplexType
@@ -890,6 +891,48 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp
     }
   }
 
+  test("array literal of empty struct falls back instead of crashing planning") {
+    // `array(array(struct()))` constant-folds to a literal whose (doubly-nested) array element
+    // type is an empty struct. `makeListLiteral` has no case for StructType at all (empty or
+    // not) -- letting this reach the literal path throws a bare `scala.MatchError` at plan
+    // time rather than a graceful fallback. See CometLiteral.isListLiteralElementSupported.
+    withTempDir { dir =>
+      val path = new Path(dir.toURI.toString, "test.parquet")
+      spark.range(10).write.parquet(path.toString)
+      withTempView("t1") {
+        spark.read.parquet(path.toString).createOrReplaceTempView("t1")
+        checkSparkAnswer(sql("SELECT id, array(array(struct())) FROM t1"))
+      }
+    }
+  }
+
+  test("array_max/array_min decline an empty-struct element type") {
+    // 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.
+    withSQLConf(
+      CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+      // Otherwise Catalyst's ConvertToLocalRelation rule evaluates this deterministic,
+      // aggregate-free projection directly over the LocalRelation's rows at plan time,
+      // producing a plan with no ArrayMax expression left to convert (or reject).
+      SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
+        "org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation") {
+      val elemType = StructType(Seq(StructField("marker", StructType(Nil))))
+      val schema =
+        StructType(Seq(StructField("id", IntegerType), StructField("a", ArrayType(elemType))))
+      val data = Seq(Row(1, Array(Row(null), Row(Row())))).asJava
+      val df = spark.createDataFrame(data, schema)
+      df.createOrReplaceTempView("empty_struct_array_extrema")
+
+      checkSparkAnswerAndFallbackReason(
+        "SELECT array_max(a) FROM empty_struct_array_extrema",
+        "array_max on a schema containing an empty struct is not supported")
+      checkSparkAnswerAndFallbackReason(
+        "SELECT array_min(a) FROM empty_struct_array_extrema",
+        "array_min on a schema containing an empty struct is not supported")
+    }
+  }
+
   test("array_reverse") {
     withTempDir { dir =>
       val path = new Path(dir.toURI.toString, "test.parquet")
diff --git a/spark/src/test/scala/org/apache/comet/CometJsonExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometJsonExpressionSuite.scala
index 2356a985ef7..84fe6ce5834 100644
--- a/spark/src/test/scala/org/apache/comet/CometJsonExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometJsonExpressionSuite.scala
@@ -166,6 +166,51 @@ class CometJsonExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe
     }
   }
 
+  test("from_json - empty struct schema") {
+    // A zero-field target schema is a legitimate value: no fields to parse into, but a real
+    // (possibly null) struct row per input, same as any other from_json result. Blank input
+    // (empty or whitespace only) is NULL per Spark's own contract (SPARK-19543), distinct from
+    // non-blank malformed input, which PERMISSIVE mode turns into a non-null struct.
+    Seq(true, false).foreach { dictionaryEnabled =>
+      withParquetTable(
+        Seq(
+          (1, "{}"), // valid JSON object
+          (2, ""), // blank input -> NULL
+          (3, "   "), // JSON-whitespace-only input (spaces) -> NULL
+          (4, "not json"), // malformed, non-blank -> non-null struct
+          (5, null), // SQL NULL input -> NULL
+          // Spark's blank check is Jackson's tokenizer finding no first token, which skips
+          // only JSON whitespace (RFC 8259: space/tab/CR/LF). Neither of the next two chars
+          // qualifies, so both must fail to tokenize -> non-null struct, not NULL.
+          (6, "\u00A0"), // non-breaking space -> not JSON whitespace, non-null struct
+          (7, "\u000B") // vertical tab -> ASCII control, not JSON whitespace, non-null struct
+        ),
+        "tbl",
+        withDictionary = dictionaryEnabled) {
+
+        checkSparkAnswerAndOperator("SELECT _1, from_json(_2, 'struct<>') FROM tbl ORDER BY _1")
+        checkSparkAnswerAndOperator(
+          "SELECT _1, from_json(_2, 'struct<>') IS NULL FROM tbl ORDER BY _1")
+      }
+    }
+  }
+
+  test("from_json - nested empty struct schema") {
+    // The zero-field struct can also appear nested inside a non-empty outer struct, which
+    // builds the result through a different code path (the nested-field builder, not the
+    // top-level one) -- both need to construct the Arrow array correctly.
+    Seq(true, false).foreach { dictionaryEnabled =>
+      withParquetTable(
+        (0 until 20).map(i => (i, """{"outer":{}}""")),
+        "tbl",
+        withDictionary = dictionaryEnabled) {
+
+        checkSparkAnswerAndOperator("SELECT from_json(_2, 'outer struct<>') FROM tbl")
+        checkSparkAnswerAndOperator("SELECT from_json(_2, 'outer struct<>').outer FROM tbl")
+      }
+    }
+  }
+
   test("from_json - nested struct") {
     Seq(true, false).foreach { dictionaryEnabled =>
       withParquetTable(
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
index 2aaf3eb8290..684c466afd8 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
@@ -107,6 +107,44 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("collect_set declines an empty-struct input") {
+    // 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.
+    withSQLConf(CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+      import scala.jdk.CollectionConverters._
+
+      import org.apache.spark.sql.functions.collect_set
+      val schema = StructType(
+        Seq(StructField("id", DataTypes.IntegerType), StructField("marker", StructType(Nil))))
+      val data = (0 until 3).map(i => Row(i, Row())).asJava
+      val df = spark.createDataFrame(data, schema)
+      checkSparkAnswer(df.agg(collect_set(col("marker"))))
+    }
+  }
+
+  test("grouping on an empty struct falls back to Spark") {
+    // DataFusion's `GroupValuesRows::emit` dictionary-encodes struct-typed group keys via
+    // `StructArray::try_new`, which errors for a zero-field struct -- see
+    // SupportLevel.containsEmptyStruct.
+    withSQLConf(CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+      import scala.jdk.CollectionConverters._
+
+      val schema = StructType(
+        Seq(StructField("id", DataTypes.IntegerType), StructField("marker", StructType(Nil))))
+      val data = (0 until 3).map(i => Row(i, Row())).asJava
+      val df = spark.createDataFrame(data, schema)
+      df.createOrReplaceTempView("empty_struct_group")
+
+      checkSparkAnswerAndFallbackReason(
+        "SELECT marker, COUNT(id) FROM empty_struct_group GROUP BY marker",
+        "Grouping on a schema containing an empty struct is not supported")
+      checkSparkAnswerAndFallbackReason(
+        "SELECT DISTINCT marker FROM empty_struct_group",
+        "Grouping on a schema containing an empty struct is not supported")
+    }
+  }
+
   test("min/max floating point with negative zero") {
     val r = new Random(42)
     val schema = StructType(
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
index bf354356386..a8c229e2420 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
@@ -21,6 +21,7 @@ package org.apache.comet.exec
 
 import java.nio.file.{Files, Paths}
 
+import scala.jdk.CollectionConverters._
 import scala.reflect.runtime.universe._
 
 import org.scalactic.source.Position
@@ -141,6 +142,21 @@ abstract class CometColumnarShuffleSuite extends CometTestBase with AdaptiveSpar
     }
   }
 
+  test("columnar shuffle on empty struct") {
+    // A struct with zero fields is a legitimate Arrow value (e.g. Iceberg's `_partition`
+    // metadata column on an unpartitioned table), not a reason to fall back to Spark. The
+    // shuffle's input must itself be Comet-native for this to exercise the real path -- a bare
+    // `LocalRelation` needs COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED to get a native upstream.
+    withSQLConf(CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+      val schema =
+        StructType(Seq(StructField("id", IntegerType), StructField("marker", StructType(Nil))))
+      val data = (0 until 50).map(i => Row(i, Row())).asJava
+      val df = spark.createDataFrame(data, schema)
+      val shuffled = df.repartition(10, $"id")
+      checkShuffleAnswer(shuffled, 1)
+    }
+  }
+
   test("columnar shuffle on array/struct map key/value") {
     // Spark 4.0 normalizes maps used as shuffle keys with mapsort(...). Comet's map_sort
     // relies on Arrow's sort_to_indices, which only supports scalar key types, so a map
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
index cd52f579559..f6bf67e2060 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
@@ -20,6 +20,7 @@
 package org.apache.comet.exec
 
 import scala.concurrent.duration.DurationInt
+import scala.jdk.CollectionConverters._
 import scala.util.Random
 
 import org.scalactic.source.Position
@@ -31,6 +32,7 @@ import org.apache.spark.sql.{CometTestBase, DataFrame, Dataset, Row}
 import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.functions.{col, count, sum}
+import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
 
 import org.apache.comet.CometConf
 
@@ -108,6 +110,21 @@ class CometNativeShuffleSuite extends CometTestBase with AdaptiveSparkPlanHelper
     }
   }
 
+  test("native shuffle on empty struct") {
+    // A struct with zero fields is a legitimate Arrow value (e.g. Iceberg's `_partition`
+    // metadata column on an unpartitioned table), not a reason to fall back to Spark. The
+    // shuffle's input must itself be Comet-native for this to exercise the real path -- a bare
+    // `LocalRelation` needs COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED to get a native upstream.
+    withSQLConf(CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+      val schema =
+        StructType(Seq(StructField("id", IntegerType), StructField("marker", StructType(Nil))))
+      val data = (0 until 50).map(i => Row(i, Row())).asJava
+      val df = spark.createDataFrame(data, schema)
+      val shuffled = df.repartition(10, $"id")
+      checkShuffleAnswer(shuffled, 1, checkNativeOperators = true)
+    }
+  }
+
   test("native shuffle over a multi-partition native scan re-threads per-partition plan data") {
     // End-to-end companion to CometNativeShuffleInputRDDSuite: that suite proves the per-partition
     // scan plan data no longer rides the broadcast task binary; this one proves each task still
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
index 7cb6816298d..0c7fa02eeee 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
@@ -19,6 +19,7 @@
 
 package org.apache.comet.exec
 
+import scala.jdk.CollectionConverters._
 import scala.util.Random
 
 import org.scalactic.source.Position
@@ -33,7 +34,7 @@ import org.apache.spark.sql.execution.window.{WindowExec => SparkWindowExec}
 import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions.{count, lead, sum}
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.DecimalType
+import org.apache.spark.sql.types.{ArrayType, DecimalType, IntegerType, StructField, StructType}
 
 import org.apache.comet.CometConf
 import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus
@@ -925,6 +926,67 @@ class CometWindowExecSuite extends CometTestBase {
     }
   }
 
+  test("window: FIRST_VALUE/LAST_VALUE decline an empty-struct input") {
+    // DataFusion's ScalarValue::compact panics reconstructing a zero-field StructArray, so
+    // CometFirst/CometLast must decline this schema and let the window run on Spark instead
+    // of letting CometWindowExec crash -- see SupportLevel.containsEmptyStruct.
+    val schema =
+      StructType(Seq(StructField("id", IntegerType), StructField("marker", StructType(Nil))))
+    val data = (0 until 3).map(i => Row(i, Row())).asJava
+    spark.createDataFrame(data, schema).createOrReplaceTempView("empty_struct_window")
+
+    checkSparkAnswer(sql("SELECT first_value(marker) OVER () FROM empty_struct_window"))
+    checkSparkAnswer(sql("SELECT last_value(marker) OVER () FROM empty_struct_window"))
+    // NTH_VALUE goes through a different DataFusion code path (a built-in window function,
+    // not an AggregateExpression accumulator) and doesn't hit ScalarValue::compact, so it
+    // needs no equivalent guard -- pinned here so a future regression would be caught.
+    checkSparkAnswer(
+      sql("SELECT nth_value(marker, 1) OVER (ORDER BY id) FROM empty_struct_window"))
+  }
+
+  test("window: LAG/LEAD decline a typed-NULL empty-struct-array default") {
+    // DataFusion casts the literal default to the input type when building the window expr,
+    // and its cast errors on a zero-field struct even when source and target agree ("no field
+    // name overlap") -- see SupportLevel.containsEmptyStruct.
+    val schema = StructType(
+      Seq(StructField("id", IntegerType), StructField("arr", ArrayType(StructType(Nil)))))
+    val data = (0 until 3).map(i => Row(i, Array(Row()))).asJava
+    spark.createDataFrame(data, schema).createOrReplaceTempView("empty_struct_array_window")
+
+    checkSparkAnswer(sql("""
+      SELECT lag(arr, 1, CAST(NULL AS ARRAY<STRUCT<>>)) OVER (ORDER BY id)
+      FROM empty_struct_array_window
+      """))
+    checkSparkAnswer(sql("""
+      SELECT lead(arr, 1, CAST(NULL AS ARRAY<STRUCT<>>)) OVER (ORDER BY id)
+      FROM empty_struct_array_window
+      """))
+
+    // The omitted-default and plain-NULL-literal forms don't go through the typed-NULL cast
+    // that panics -- they should stay native, not fall back with the rest of this schema.
+    checkSparkAnswerAndOperator(
+      sql("SELECT lag(arr, 1) OVER (ORDER BY id) FROM empty_struct_array_window"))
+    checkSparkAnswerAndOperator(
+      sql("SELECT lag(arr, 1, NULL) OVER (ORDER BY id) FROM empty_struct_array_window"))
+  }
+
+  test("window: RANGE frame declines an empty-struct ORDER BY key") {
+    // DataFusion's `ScalarValue::partial_cmp_struct` flattens a struct into its leaf columns to
+    // compare RANGE peers; a zero-field struct contributes no columns, so a NULL struct and a
+    // non-null empty struct compare equal instead of ordering NULL first -- misgrouping peers.
+    val schema = StructType(
+      Seq(
+        StructField("s", StructType(Seq(StructField("e", StructType(Nil))))),
+        StructField("k", IntegerType)))
+    val data = Seq(Row(Row(null), 1), Row(Row(Row()), 1), Row(Row(null), 2)).asJava
+    spark.createDataFrame(data, schema).createOrReplaceTempView("empty_struct_range_window")
+
+    checkSparkAnswerAndFallbackReason(
+      "SELECT COUNT(*) OVER (ORDER BY s, k RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) " +
+        "FROM empty_struct_range_window",
+      "RANGE frame ordering on a schema containing an empty struct is not supported")
+  }
+
   test("window: LAST_VALUE with ROWS frame") {
     withTempDir { dir =>
       (0 until 30)