From de987e83d1765a521f487f22e09a81a6d58b6321 Mon Sep 17 00:00:00 2001
From: comphead <comphead@ukr.net>
Date: Tue, 28 Jul 2026 09:01:00 -0700
Subject: [PATCH 1/9] feat: route CometCast through codegen dispatcher for
 legacy config paths

Mix CodegenDispatchFallback into CometCast so casts that Comet cannot
match Spark bit-for-bit run Spark's own doGenCode inside the Comet
pipeline instead of falling the full projection back to Spark:

- spark.sql.legacy.castComplexTypesToString.enabled=true (struct/array/map
  to string) is now Incompatible, was Unsupported which forced full fallback.
- Non-CORRECTED spark.sql.legacy.timeParserPolicy (LEGACY, EXCEPTION) on
  string to date/timestamp casts is now Incompatible.
- Pre-existing Incompatible cases in CometCast (negative-scale decimal
  to string, float/double to decimal rounding) also start using the codegen
  dispatcher through the mixin.

VariantType casts remain Unsupported and reach the dispatcher too. The
mixin surfaces whatever error Spark itself would emit for Variant.
isVariantType is added to CometTypeShim for the guard.

Tests:
- cast_complex_types_to_string_legacy.sql: drop expect_fallback since
  these casts now stay in the Comet pipeline.
- New cast_string_to_date_time_parser_policy_legacy.sql: exercise
  string to date/timestamp/timestamp_ntz casts under LEGACY timeParserPolicy.
- CometCastSuite: cast ArrayType(DateType) to unsupported ArrayType now
  checks the Comet operator rather than a Spark fallback reason.
---
 .../apache/comet/expressions/CometCast.scala  | 56 ++++++++++++++-----
 .../apache/comet/shims/CometTypeShim.scala    |  3 +
 .../apache/comet/shims/CometTypeShim.scala    |  8 ++-
 .../cast_complex_types_to_string_legacy.sql   | 22 ++++----
 ...ring_to_date_time_parser_policy_legacy.sql | 50 +++++++++++++++++
 .../org/apache/comet/CometCastSuite.scala     | 17 +-----
 6 files changed, 116 insertions(+), 40 deletions(-)
 create mode 100644 spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql

diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index 619b69912f..3484651ece 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -25,30 +25,41 @@ import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DecimalType,
 
 import org.apache.comet.CometConf
 import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, withFallbackReason}
-import org.apache.comet.serde.{CometExpressionSerde, Compatible, ExprOuterClass, Incompatible, SupportLevel, Unsupported}
+import org.apache.comet.DataTypeSupport.isComplexType
+import org.apache.comet.serde.{CodegenDispatchFallback, CometExpressionSerde, Compatible, ExprOuterClass, Incompatible, SupportLevel, Unsupported}
 import org.apache.comet.serde.ExprOuterClass.Expr
 import org.apache.comet.serde.QueryPlanSerde.{evalModeToProto, exprToProtoInternal, serializeDataType}
-import org.apache.comet.shims.CometExprShim
+import org.apache.comet.shims.{CometExprShim, CometTypeShim}
 
-object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
+object CometCast
+    extends CometExpressionSerde[Cast]
+    with CometExprShim
+    with CometTypeShim
+    with CodegenDispatchFallback {
 
   // Shared with CometCastSuite so the asserted reason cannot drift from production.
   private[comet] val negativeScaleDecimalToStringReason: String =
     "Negative-scale decimal requires spark.sql.legacy.allowNegativeScaleOfDecimal=true"
 
-  // When `spark.sql.legacy.castComplexTypesToString.enabled` is true, Spark wraps maps and
-  // structs with `[]` (instead of `{}`) when casting to string, and omits NULL elements of
-  // structs/maps/arrays (instead of rendering them as the literal "null"). Comet only
-  // implements the default formatting, so fall back to Spark for any array/map/struct to-string
-  // cast when the flag is enabled. The flag is internal in Spark 4.0 and defaults to false.
   private[comet] val legacyCastComplexTypesToStringReason: String =
-    "spark.sql.legacy.castComplexTypesToString.enabled=true is not supported"
+    "spark.sql.legacy.castComplexTypesToString.enabled=true is not supported natively"
+
+  private[comet] val nonDefaultTimeParserPolicyReason: String =
+    "spark.sql.legacy.timeParserPolicy is set to a non-CORRECTED value; the native " +
+      "string-to-datetime parser only implements CORRECTED semantics"
 
   private def legacyCastComplexTypesToString: Boolean =
     SQLConf.get
       .getConfString("spark.sql.legacy.castComplexTypesToString.enabled", "false")
       .toBoolean
 
+  // Non-CORRECTED policies (LEGACY, EXCEPTION) change string-to-date/timestamp parsing behavior in
+  // ways the native cast kernel does not replicate.
+  private def isNonDefaultTimeParserPolicy: Boolean =
+    !SQLConf.get
+      .getConfString("spark.sql.legacy.timeParserPolicy", "CORRECTED")
+      .equalsIgnoreCase("CORRECTED")
+
   def supportedTypes: Seq[DataType] =
     Seq(
       DataTypes.BooleanType,
@@ -72,6 +83,14 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
   // would only duplicate the matrix and risk drifting from it.
 
   override def getSupportLevel(cast: Cast): SupportLevel = {
+    // Reject `VariantType` before the Literal short-circuit below. Folding a Cast whose child or
+    // target is `VariantType` produces a `Literal[VariantType]` that no downstream Comet serde
+    // can serialize, and relying on `CometLiteral` to reject it after the fact leaves a native
+    // path that assumes the produced literal is safe. Guarding here (in addition to the
+    // recursive check in `isSupported`) forces Spark fallback for every VariantType cast shape.
+    if (isVariantType(cast.child.dataType) || isVariantType(cast.dataType)) {
+      return unsupported(cast.child.dataType, cast.dataType)
+    }
     if (cast.child.isInstanceOf[Literal]) {
       // A cast whose child is a literal is folded by Spark at planning time via `cast.eval()`
       // (see `convert`), so the cast never executes natively and the result matches Spark by
@@ -159,14 +178,22 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
       timeZoneId: Option[String],
       evalMode: CometEvalMode.Value): SupportLevel = {
 
+    // Spark 4's `VariantType` (SPARK-45827) has no native counterpart in Comet: serializing it
+    // into the DataFusion plan would fail in `serializeDataType`, and the codegen dispatcher
+    // cannot compile Variant read/write kernels either. Detect it via the version-shimmed
+    // `isVariantType` (which returns false on Spark 3.x where the class does not exist) and
+    // report `Unsupported` so the enclosing operator falls back to Spark.
+    if (isVariantType(fromType) || isVariantType(toType)) {
+      return unsupported(fromType, toType)
+    }
+
     if (fromType == toType) {
       return Compatible()
     }
 
-    if (toType == DataTypes.StringType && legacyCastComplexTypesToString && (fromType
-        .isInstanceOf[ArrayType] || fromType.isInstanceOf[StructType] ||
-        fromType.isInstanceOf[MapType])) {
-      return Unsupported(Some(legacyCastComplexTypesToStringReason))
+    if (toType == DataTypes.StringType && legacyCastComplexTypesToString && isComplexType(
+        fromType)) {
+      return Incompatible(Some(legacyCastComplexTypesToStringReason))
     }
 
     (fromType, toType) match {
@@ -248,6 +275,9 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
         Compatible()
       case _: DecimalType =>
         Compatible()
+      case DataTypes.DateType | DataTypes.TimestampType | _: TimestampNTZType
+          if isNonDefaultTimeParserPolicy =>
+        Incompatible(Some(nonDefaultTimeParserPolicyReason))
       case DataTypes.DateType =>
         // https://github.com/apache/datafusion-comet/issues/327
         Compatible(Some("Only supports years between 262143 BC and 262142 AD"))
diff --git a/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala b/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala
index 97320be9e7..b71476c3dd 100644
--- a/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala
+++ b/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala
@@ -36,6 +36,9 @@ trait CometTypeShim {
   @nowarn // Spark 4 feature; Variant shredding doesn't exist in Spark 3.x.
   def isVariantStruct(s: StructType): Boolean = false
 
+  @nowarn // Spark 4 feature; VariantType doesn't exist in Spark 3.x.
+  def isVariantType(dt: DataType): Boolean = false
+
   @nowarn // Spark 4.1 feature; TimeType doesn't exist in Spark 3.x.
   def isTimeType(dt: DataType): Boolean = false
 }
diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala
index 1d4a9f601e..f48955a7da 100644
--- a/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala
+++ b/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala
@@ -20,7 +20,7 @@
 package org.apache.comet.shims
 
 import org.apache.spark.sql.execution.datasources.VariantMetadata
-import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType, StructType}
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType, StructType, VariantType}
 
 trait CometTypeShim {
   // A `StringType` carries collation metadata in Spark 4.0. Only non-default (non-UTF8_BINARY)
@@ -54,6 +54,12 @@ trait CometTypeShim {
   // and force scan fallback.
   def isVariantStruct(s: StructType): Boolean = VariantMetadata.isVariantStruct(s)
 
+  // Comet has no native execution path for Spark 4's `VariantType` (introduced in
+  // SPARK-45827). Serdes call this to route casts/expressions touching the type back to Spark
+  // rather than serializing an unsupported datatype into the native plan. Stubbed to `false` in
+  // Spark 3.x where `VariantType` does not exist.
+  def isVariantType(dt: DataType): Boolean = dt.isInstanceOf[VariantType]
+
   def isTimeType(dt: DataType): Boolean =
     dt.getClass.getSimpleName.startsWith("TimeType")
 
diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql
index 2c0bc19b3b..92010a7c89 100644
--- a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql
+++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql
@@ -17,24 +17,26 @@
 
 -- When `spark.sql.legacy.castComplexTypesToString.enabled` is true Spark wraps maps and
 -- structs with `[...]` (instead of `{...}`) and omits NULL elements of structs/maps/arrays
--- (instead of rendering them as the literal "null"). Comet only implements the default
--- formatting, so any array/map/struct → string cast must fall back to Spark.
+-- (instead of rendering them as the literal "null"). Comet's native cast does not implement
+-- the legacy formatting; the [[CodegenDispatchFallback]] mixin on `CometCast` routes these
+-- casts through the JVM codegen dispatcher (Spark's own `doGenCode` inside the Comet kernel)
+-- so results match Spark exactly without a Spark fallback.
 -- The flag is internal in Spark 4.0 and defaults to false.
 
 -- Config: spark.sql.legacy.castComplexTypesToString.enabled=true
 
--- Struct → string falls back.
-query expect_fallback(spark.sql.legacy.castComplexTypesToString.enabled=true is not supported)
+-- Struct → string routed through the codegen dispatcher.
+query
 SELECT CAST(struct(1, 2, null) AS STRING)
 
--- Array → string falls back (NULL elements rendered differently between modes).
-query expect_fallback(spark.sql.legacy.castComplexTypesToString.enabled=true is not supported)
+-- Array → string routed through the codegen dispatcher.
+query
 SELECT CAST(array(1, 2, null) AS STRING)
 
--- Map → string falls back (`[]` vs `{}` wrapping differs between modes).
-query expect_fallback(spark.sql.legacy.castComplexTypesToString.enabled=true is not supported)
+-- Map → string routed through the codegen dispatcher.
+query
 SELECT CAST(map('a', 1, 'b', null) AS STRING)
 
--- Nested complex types still fall back through the outer type.
-query expect_fallback(spark.sql.legacy.castComplexTypesToString.enabled=true is not supported)
+-- Nested complex types also routed through the codegen dispatcher via the outer type.
+query
 SELECT CAST(struct(array(1, null), map('k', null)) AS STRING)
diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql
new file mode 100644
index 0000000000..1a966db985
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql
@@ -0,0 +1,50 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License.  You may obtain a copy of the License at
+--
+--   http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied.  See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- CAST(string AS date/timestamp) under LEGACY timeParserPolicy.
+-- LEGACY / EXCEPTION policies use `SimpleDateFormat` semantics, which the native
+-- string-to-datetime kernel does not replicate. `CometCast` reports `Incompatible`
+-- and the `CodegenDispatchFallback` mixin routes the cast through the JVM codegen
+-- dispatcher (Spark's own `doGenCode` inside the Comet pipeline), so results match
+-- Spark exactly without a full Spark fallback.
+-- Config: spark.sql.legacy.timeParserPolicy=LEGACY
+-- Config: spark.sql.session.timeZone=UTC
+
+statement
+CREATE TABLE test_cast_string_dt(s string) USING parquet
+
+statement
+INSERT INTO test_cast_string_dt VALUES
+  ('2024-01-01'),
+  ('2024-1-1'),
+  ('2024-13-01'),
+  ('2024-02-30'),
+  ('2024-01-01garbage'),
+  ('2024'),
+  (NULL)
+
+-- Cast to DateType under LEGACY timeParserPolicy.
+query spark_answer_only
+SELECT s, CAST(s AS DATE) FROM test_cast_string_dt ORDER BY s
+
+-- Cast to TimestampType under LEGACY timeParserPolicy.
+query spark_answer_only
+SELECT s, CAST(s AS TIMESTAMP) FROM test_cast_string_dt ORDER BY s
+
+-- Cast to TimestampNTZType under LEGACY timeParserPolicy.
+query spark_answer_only
+SELECT s, CAST(s AS TIMESTAMP_NTZ) FROM test_cast_string_dt ORDER BY s
diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
index abc04bfbb5..daa85c4f92 100644
--- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
@@ -735,19 +735,6 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("cast DecimalType with negative scale to StringType") {
-    // Negative-scale decimals are a legacy Spark feature gated on
-    // spark.sql.legacy.allowNegativeScaleOfDecimal=true. Spark LEGACY cast uses Java's
-    // BigDecimal.toString() which produces scientific notation for negative-scale values
-    // (e.g. 12300 stored as Decimal(7,-2) with unscaled=123 → "1.23E+4").
-    // CometCast.canCastToString checks the
-    // config and returns Incompatible when it is false.
-    //
-    // Parquet does not support negative-scale decimals so we use checkSparkAnswer directly
-    // (no parquet round-trip) to avoid schema coercion.
-
-    // With config enabled, enable localTableScan so Comet can take over the full plan
-    // and execute the cast natively. Parquet does not support negative-scale decimals so
-    // the data is kept in-memory; localTableScan.enabled bridges that gap.
     withSQLConf(
       "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true",
       "spark.comet.exec.localTableScan.enabled" -> "true") {
@@ -1814,9 +1801,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
           assert(
             CometCast.isSupported(fromType, toType, None, CometEvalMode.LEGACY) ==
               Unsupported(Some(expectedMessage)))
-          checkSparkAnswerAndFallbackReason(
-            data.select(col("a").cast(toType).as("converted")),
-            expectedMessage)
+          checkSparkAnswerAndOperator(data.select(col("a").cast(toType).as("converted")))
         }
       }
     }

From 7bac23078beb432e270d97d4b5900d79ccfe5410 Mon Sep 17 00:00:00 2001
From: comphead <comphead@ukr.net>
Date: Tue, 28 Jul 2026 09:11:17 -0700
Subject: [PATCH 2/9] test: drop expect_fallback markers from non-legacy
 cast_complex_types_to_string

Map to string and struct/array containing Map to string casts are Unsupported
in CometCast, but with the CodegenDispatchFallback mixin they now run through
Spark's own doGenCode inside the Comet pipeline. The Comet operator stays in
the plan, so expect_fallback assertions no longer hold.
---
 .../cast/cast_complex_types_to_string.sql     | 35 +++++++++----------
 1 file changed, 16 insertions(+), 19 deletions(-)

diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
index 8b1d989ae7..ac05b67675 100644
--- a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
+++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
@@ -149,8 +149,7 @@ SELECT cast(named_struct('a', named_struct('b', named_struct('c', 1, 'd', 'leaf'
 query
 SELECT cast(named_struct('s1', '', 's2', ' ', 's3', cast(null as string)) as string)
 
--- Map-valued field: not supported, falls back to Spark.
-query expect_fallback(to StringType is not supported)
+query
 SELECT cast(named_struct('m', map('k', 1)) as string)
 
 -- ----------------------------------------------------------------------------
@@ -270,8 +269,6 @@ SELECT cast(array(cast(1.5 as double), cast('NaN' as double), cast('-Infinity' a
 query
 SELECT cast(array(array(array(1, 2), array(3)), array(array(cast(null as int)))) as string)
 
--- Array of map: not supported, falls back to Spark.
-query expect_fallback(to StringType is not supported)
 SELECT cast(array(map('k', 1)) as string)
 
 -- ----------------------------------------------------------------------------
@@ -282,57 +279,57 @@ SELECT cast(array(map('k', 1)) as string)
 -- tests use literal maps directly rather than reading from a parquet table.
 
 -- Map with string keys, int values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', 1, 'b', 2, 'c', 3) as string)
 
 -- Map with NULL values rendered as "null".
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', 1, 'b', cast(null as int), 'c', 3) as string)
 
 -- Map with int keys, string values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map(1, 'one', 2, 'two', 3, 'three') as string)
 
 -- Map with boolean values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('t', true, 'f', false, 'n', cast(null as boolean)) as string)
 
 -- Map with bigint values at min/max.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('max', 9223372036854775807, 'min', -9223372036854775808, 'zero', cast(0 as bigint)) as string)
 
 -- Map with decimal values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('pos', cast('1.234567890123456789' as decimal(38, 18)), 'neg', cast('-1.234567890123456789' as decimal(38, 18)), 'null', cast(null as decimal(38, 18))) as string)
 
 -- Map with date and timestamp values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', date '2024-01-15', 'b', date '1970-01-01', 'c', cast(null as date)) as string)
 
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', timestamp '2024-01-15 10:30:45', 'b', cast(null as timestamp)) as string)
 
 -- Map with binary values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', X'616263', 'b', X'', 'c', cast(null as binary)) as string)
 
 -- Map with float / double values: NaN / ±0 / ±Infinity / NULL.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('nan', cast('NaN' as float), 'neg0', cast(-0.0 as float), 'null', cast(null as float)) as string)
 
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('nan', cast('NaN' as double), 'inf', cast('Infinity' as double), 'ninf', cast('-Infinity' as double), 'null', cast(null as double)) as string)
 
 -- Map with struct values: each value rendered as `{f1, f2, ...}`.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', named_struct('x', 1, 'y', 'first'), 'b', cast(null as struct<x: int, y: string>)) as string)
 
 -- Map with array values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', array(1, 2, 3), 'b', array(cast(null as int)), 'c', cast(null as array<int>)) as string)
 
 -- Empty map.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map() as string)
 
 -- NULL map: Spark constant-folds this to a literal NULL, so the cast never reaches Comet
@@ -341,5 +338,5 @@ query
 SELECT cast(cast(null as map<string, int>) as string)
 
 -- Map of map.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('outer', map('inner', 1)) as string)

From 9f0fc10549fe4dbccd48d8f1e43701058a74b8bd Mon Sep 17 00:00:00 2001
From: comphead <comphead@ukr.net>
Date: Tue, 28 Jul 2026 09:12:13 -0700
Subject: [PATCH 3/9] fix: restore query directive and refresh Map header in
 cast_complex_types_to_string

- Add a query directive above `SELECT cast(array(map('k', 1)) as string)` so the
  parser actually runs the array-of-map case (previously an orphan SELECT).
- Rewrite the Map to string section header to describe codegen dispatch behavior.
---
 .../expressions/cast/cast_complex_types_to_string.sql       | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
index ac05b67675..cf16f9f8f5 100644
--- a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
+++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
@@ -269,12 +269,16 @@ SELECT cast(array(cast(1.5 as double), cast('NaN' as double), cast('-Infinity' a
 query
 SELECT cast(array(array(array(1, 2), array(3)), array(array(cast(null as int)))) as string)
 
+-- Array of map: map-to-string is routed through the codegen dispatcher via the outer array.
+query
 SELECT cast(array(map('k', 1)) as string)
 
 -- ----------------------------------------------------------------------------
 -- Map → string
 -- ----------------------------------------------------------------------------
--- Comet does not implement map-to-string casts, so every map → string falls back to Spark.
+-- Comet does not implement map-to-string casts natively; the CodegenDispatchFallback
+-- mixin on CometCast routes them through Spark's doGenCode inside the Comet pipeline,
+-- so results match Spark exactly without a full Spark fallback.
 -- Note: maps materialized through parquet have nondeterministic entry order, so map column
 -- tests use literal maps directly rather than reading from a parquet table.
 

From bd237d60074f78149c5784f03ee0898ab062a0b1 Mon Sep 17 00:00:00 2001
From: comphead <comphead@ukr.net>
Date: Tue, 28 Jul 2026 09:17:50 -0700
Subject: [PATCH 4/9] test: run cast_string_to_date_time_parser_policy_legacy
 with plain query mode

Under LEGACY timeParserPolicy the CodegenDispatchFallback mixin keeps the
CAST inside the Comet pipeline via Spark's doGenCode, so the Comet operator
should be present in the plan. Use plain `query` (not `spark_answer_only`)
so the test verifies both the result and that Comet is executing the cast.
---
 .../cast/cast_string_to_date_time_parser_policy_legacy.sql  | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql
index 1a966db985..2f45066392 100644
--- a/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql
+++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql
@@ -38,13 +38,13 @@ INSERT INTO test_cast_string_dt VALUES
   (NULL)
 
 -- Cast to DateType under LEGACY timeParserPolicy.
-query spark_answer_only
+query
 SELECT s, CAST(s AS DATE) FROM test_cast_string_dt ORDER BY s
 
 -- Cast to TimestampType under LEGACY timeParserPolicy.
-query spark_answer_only
+query
 SELECT s, CAST(s AS TIMESTAMP) FROM test_cast_string_dt ORDER BY s
 
 -- Cast to TimestampNTZType under LEGACY timeParserPolicy.
-query spark_answer_only
+query
 SELECT s, CAST(s AS TIMESTAMP_NTZ) FROM test_cast_string_dt ORDER BY s

From b03497ffb62f2a255cf5574d8a39a6d32200bd79 Mon Sep 17 00:00:00 2001
From: comphead <comphead@ukr.net>
Date: Tue, 28 Jul 2026 11:14:51 -0700
Subject: [PATCH 5/9] test: update explain-comet expectations for
 CalendarInterval cast via codegen dispatch

The `explain comet` test asserted that `cast(make_interval(...) as string)` falls
back to Spark with reason `Cast from CalendarIntervalType to StringType is not
supported`. With the CodegenDispatchFallback mixin on CometCast, that cast now
runs through Spark's doGenCode inside the Comet pipeline, so the operator stays
Comet-native and no fallback reason is emitted:
- Drop the standalone `SELECT cast(make_interval(...) as string)` case whose
  only expectation was that reason.
- Prune the same reason from the join case; the shuffle-disabled reason still
  applies to the aggregate side of the join.
---
 .../test/scala/org/apache/comet/CometExpressionSuite.scala | 7 +------
 1 file changed, 1 insertion(+), 6 deletions(-)

diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
index b8a70760bf..5af03be9d1 100644
--- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
@@ -2034,9 +2034,6 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
         sql(s"insert into $table values(0, 1, 100.000001)")
 
         Seq(
-          (
-            s"SELECT cast(make_interval(c0, c1, c0, c1, c0, c0, c2) as string) as C from $table",
-            Set("Cast from CalendarIntervalType to StringType is not supported")),
           (
             "SELECT "
               + "date_part('YEAR', make_interval(c0, c1, c0, c1, c0, c0, c2))"
@@ -2054,9 +2051,7 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
               + s"(SELECT c1, sum(c0) as sum_c0, sum(c2) as sum_c2 from $table group by c1) as A, "
               + s"(SELECT c1, cast(make_interval(c0, c1, c0, c1, c0, c0, c2) as string) as casted from $table) as B "
               + "where A.c1 = B.c1 ",
-            Set(
-              "Cast from CalendarIntervalType to StringType is not supported",
-              "Comet shuffle is not enabled: spark.comet.shuffle.enabled is not enabled")),
+            Set("Comet shuffle is not enabled: spark.comet.shuffle.enabled is not enabled")),
           (s"select * from $table LIMIT 10 OFFSET 3", Set("Comet shuffle is not enabled")))
           .foreach(test => {
             val qry = test._1

From 25c3d1c43d47a4e19b10d490b286c25c9bfb6344 Mon Sep 17 00:00:00 2001
From: comphead <comphead@ukr.net>
Date: Tue, 28 Jul 2026 17:51:38 -0700
Subject: [PATCH 6/9] address comments

---
 .../apache/comet/expressions/CometCast.scala  | 30 ++++-------
 ...ring_to_date_time_parser_policy_legacy.sql | 50 -------------------
 .../org/apache/comet/CometCastSuite.scala     | 24 ++++++++-
 3 files changed, 32 insertions(+), 72 deletions(-)
 delete mode 100644 spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql

diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index 3484651ece..b4cad10d6e 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -44,22 +44,11 @@ object CometCast
   private[comet] val legacyCastComplexTypesToStringReason: String =
     "spark.sql.legacy.castComplexTypesToString.enabled=true is not supported natively"
 
-  private[comet] val nonDefaultTimeParserPolicyReason: String =
-    "spark.sql.legacy.timeParserPolicy is set to a non-CORRECTED value; the native " +
-      "string-to-datetime parser only implements CORRECTED semantics"
-
   private def legacyCastComplexTypesToString: Boolean =
     SQLConf.get
       .getConfString("spark.sql.legacy.castComplexTypesToString.enabled", "false")
       .toBoolean
 
-  // Non-CORRECTED policies (LEGACY, EXCEPTION) change string-to-date/timestamp parsing behavior in
-  // ways the native cast kernel does not replicate.
-  private def isNonDefaultTimeParserPolicy: Boolean =
-    !SQLConf.get
-      .getConfString("spark.sql.legacy.timeParserPolicy", "CORRECTED")
-      .equalsIgnoreCase("CORRECTED")
-
   def supportedTypes: Seq[DataType] =
     Seq(
       DataTypes.BooleanType,
@@ -86,8 +75,9 @@ object CometCast
     // Reject `VariantType` before the Literal short-circuit below. Folding a Cast whose child or
     // target is `VariantType` produces a `Literal[VariantType]` that no downstream Comet serde
     // can serialize, and relying on `CometLiteral` to reject it after the fact leaves a native
-    // path that assumes the produced literal is safe. Guarding here (in addition to the
-    // recursive check in `isSupported`) forces Spark fallback for every VariantType cast shape.
+    // path that assumes the produced literal is safe. Report `Unsupported`; the
+    // `CodegenDispatchFallback` mixin will then try the codegen dispatcher, which itself cannot
+    // serialize `VariantType` data args or return types, so the operator falls back to Spark.
     if (isVariantType(cast.child.dataType) || isVariantType(cast.dataType)) {
       return unsupported(cast.child.dataType, cast.dataType)
     }
@@ -179,10 +169,11 @@ object CometCast
       evalMode: CometEvalMode.Value): SupportLevel = {
 
     // Spark 4's `VariantType` (SPARK-45827) has no native counterpart in Comet: serializing it
-    // into the DataFusion plan would fail in `serializeDataType`, and the codegen dispatcher
-    // cannot compile Variant read/write kernels either. Detect it via the version-shimmed
-    // `isVariantType` (which returns false on Spark 3.x where the class does not exist) and
-    // report `Unsupported` so the enclosing operator falls back to Spark.
+    // into the DataFusion plan fails in `serializeDataType`, and the codegen dispatcher used by
+    // the `CodegenDispatchFallback` mixin also cannot serialize `VariantType` in the data args or
+    // return type. Detect it via the version-shimmed `isVariantType` (which returns false on
+    // Spark 3.x where the class does not exist) and report `Unsupported`. The mixin will attempt
+    // the codegen dispatcher and then fall back to Spark when the dispatcher rejects the type.
     if (isVariantType(fromType) || isVariantType(toType)) {
       return unsupported(fromType, toType)
     }
@@ -193,7 +184,7 @@ object CometCast
 
     if (toType == DataTypes.StringType && legacyCastComplexTypesToString && isComplexType(
         fromType)) {
-      return Incompatible(Some(legacyCastComplexTypesToStringReason))
+      return Unsupported(Some(legacyCastComplexTypesToStringReason))
     }
 
     (fromType, toType) match {
@@ -275,9 +266,6 @@ object CometCast
         Compatible()
       case _: DecimalType =>
         Compatible()
-      case DataTypes.DateType | DataTypes.TimestampType | _: TimestampNTZType
-          if isNonDefaultTimeParserPolicy =>
-        Incompatible(Some(nonDefaultTimeParserPolicyReason))
       case DataTypes.DateType =>
         // https://github.com/apache/datafusion-comet/issues/327
         Compatible(Some("Only supports years between 262143 BC and 262142 AD"))
diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql
deleted file mode 100644
index 2f45066392..0000000000
--- a/spark/src/test/resources/sql-tests/expressions/cast/cast_string_to_date_time_parser_policy_legacy.sql
+++ /dev/null
@@ -1,50 +0,0 @@
--- Licensed to the Apache Software Foundation (ASF) under one
--- or more contributor license agreements.  See the NOTICE file
--- distributed with this work for additional information
--- regarding copyright ownership.  The ASF licenses this file
--- to you under the Apache License, Version 2.0 (the
--- "License"); you may not use this file except in compliance
--- with the License.  You may obtain a copy of the License at
---
---   http://www.apache.org/licenses/LICENSE-2.0
---
--- Unless required by applicable law or agreed to in writing,
--- software distributed under the License is distributed on an
--- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
--- KIND, either express or implied.  See the License for the
--- specific language governing permissions and limitations
--- under the License.
-
--- CAST(string AS date/timestamp) under LEGACY timeParserPolicy.
--- LEGACY / EXCEPTION policies use `SimpleDateFormat` semantics, which the native
--- string-to-datetime kernel does not replicate. `CometCast` reports `Incompatible`
--- and the `CodegenDispatchFallback` mixin routes the cast through the JVM codegen
--- dispatcher (Spark's own `doGenCode` inside the Comet pipeline), so results match
--- Spark exactly without a full Spark fallback.
--- Config: spark.sql.legacy.timeParserPolicy=LEGACY
--- Config: spark.sql.session.timeZone=UTC
-
-statement
-CREATE TABLE test_cast_string_dt(s string) USING parquet
-
-statement
-INSERT INTO test_cast_string_dt VALUES
-  ('2024-01-01'),
-  ('2024-1-1'),
-  ('2024-13-01'),
-  ('2024-02-30'),
-  ('2024-01-01garbage'),
-  ('2024'),
-  (NULL)
-
--- Cast to DateType under LEGACY timeParserPolicy.
-query
-SELECT s, CAST(s AS DATE) FROM test_cast_string_dt ORDER BY s
-
--- Cast to TimestampType under LEGACY timeParserPolicy.
-query
-SELECT s, CAST(s AS TIMESTAMP) FROM test_cast_string_dt ORDER BY s
-
--- Cast to TimestampNTZType under LEGACY timeParserPolicy.
-query
-SELECT s, CAST(s AS TIMESTAMP_NTZ) FROM test_cast_string_dt ORDER BY s
diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
index daa85c4f92..e0c0299144 100644
--- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
@@ -1784,7 +1784,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
         Unsupported(Some(expectedMessage)))
   }
 
-  test("cast ArrayType(DateType) to unsupported ArrayType falls back") {
+  test("cast ArrayType(DateType) to unsupported ArrayType routes through codegen dispatch") {
     val fromType = ArrayType(DateType)
     val unsupportedElementTypes =
       Seq(BooleanType, ByteType, ShortType, LongType, FloatType, DoubleType, DecimalType(10, 2))
@@ -1855,6 +1855,28 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("cast to and from VariantType matches Spark") {
+    // Spark 4's VariantType has no native counterpart in Comet: the CometCast
+    // matrix reports Unsupported, and the codegen dispatcher cannot serialize a
+    // VariantType data arg or output type either, so the operator falls back to
+    // Spark cleanly. Guard on Spark 4.0+ because VariantType and parse_json do
+    // not exist in Spark 3.x.
+    assume(CometSparkSessionExtensions.isSpark40Plus, "VariantType requires Spark 4.0+")
+    withTable("variant_cast") {
+      sql("CREATE TABLE variant_cast(id INT, s STRING) USING parquet")
+      sql("""
+          |INSERT INTO variant_cast VALUES
+          |  (1, '{"a": 1}'),
+          |  (2, '{"b": [1, 2, 3]}'),
+          |  (3, cast(null as string))
+          """.stripMargin)
+      checkSparkAnswer(
+        "SELECT id, CAST(parse_json(s) AS STRING) AS v FROM variant_cast ORDER BY id")
+      checkSparkAnswer(
+        "SELECT id, CAST(CAST(s AS VARIANT) AS STRING) AS v FROM variant_cast ORDER BY id")
+    }
+  }
+
   private def testArrayCastMatrix(
       elementTypes: Seq[DataType],
       wrapType: DataType => DataType,

From a219801357714a58fc718e62c076d35b7cc7f41f Mon Sep 17 00:00:00 2001
From: comphead <comphead@ukr.net>
Date: Wed, 29 Jul 2026 08:10:46 -0700
Subject: [PATCH 7/9] test(spark-diff): allow Comet plan in
 VariantEndToEndSuite codegen assertion

VariantEndToEndSuite has a codegen-mode plan-type assertion that expects
WholeStageCodegenExec at the top. Under Comet the plan is wrapped in
CometNativeColumnarToRowExec, so the strict class check fails once
CometCast (with CodegenDispatchFallback) keeps the Variant cast inside
the Comet pipeline. Loosen the assertion to also accept
CometNativeColumnarToRowExec in dev/diffs/4.0.2.diff and 4.1.2.diff.

Regenerated by applying the delta to Spark v4.0.2 and v4.1.2 clones and
verifying with `git apply --check` on both.
---
 dev/diffs/4.0.2.diff | 22 ++++++++++++++++++++++
 dev/diffs/4.1.2.diff | 22 ++++++++++++++++++++++
 2 files changed, 44 insertions(+)

diff --git a/dev/diffs/4.0.2.diff b/dev/diffs/4.0.2.diff
index a2f6eac74a..0485a2d5f7 100644
--- a/dev/diffs/4.0.2.diff
+++ b/dev/diffs/4.0.2.diff
@@ -1380,6 +1380,28 @@ index 2e33f6505ab..be967565303 100644
      }
  
      withTable("t1", "t2") {
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+index a40e34d94d0..abc1f035d15 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+@@ -21,6 +21,7 @@ import org.apache.spark.sql.QueryTest.sameRows
+ import org.apache.spark.sql.catalyst.InternalRow
+ import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+ import org.apache.spark.sql.catalyst.expressions.variant.{ToVariantObject, VariantExpressionEvalUtils}
++import org.apache.spark.sql.comet.CometNativeColumnarToRowExec
+ import org.apache.spark.sql.execution.WholeStageCodegenExec
+ import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+ import org.apache.spark.sql.functions._
+@@ -358,7 +359,8 @@ class VariantEndToEndSuite extends QueryTest with SharedSparkSession {
+             s"cast(to_variant_object(s) as ${schema(2).dataType.sql})")
+           checkAnswer(df, input)
+           val plan = df.queryExecution.executedPlan
+-          assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY"))
++          assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY")
++            || plan.isInstanceOf[CometNativeColumnarToRowExec])
+         }
+       }
+     }
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
 index 11e9547dfc5..637411056ae 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
diff --git a/dev/diffs/4.1.2.diff b/dev/diffs/4.1.2.diff
index 30f6c2f23f..8bd4bfe732 100644
--- a/dev/diffs/4.1.2.diff
+++ b/dev/diffs/4.1.2.diff
@@ -1479,6 +1479,28 @@ index 3ba48da0e32..0313aaa9ec6 100644
      }
  
      withTable("t1", "t2") {
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+index 8a0e2c29653..d276a51cbc6 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+@@ -21,6 +21,7 @@ import org.apache.spark.sql.QueryTest.sameRows
+ import org.apache.spark.sql.catalyst.InternalRow
+ import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+ import org.apache.spark.sql.catalyst.expressions.variant.{ToVariantObject, VariantExpressionEvalUtils}
++import org.apache.spark.sql.comet.CometNativeColumnarToRowExec
+ import org.apache.spark.sql.execution.WholeStageCodegenExec
+ import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+ import org.apache.spark.sql.functions._
+@@ -358,7 +359,8 @@ class VariantEndToEndSuite extends QueryTest with SharedSparkSession {
+             s"cast(to_variant_object(s) as ${schema(2).dataType.sql})")
+           checkAnswer(df, input)
+           val plan = df.queryExecution.executedPlan
+-          assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY"))
++          assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY")
++            || plan.isInstanceOf[CometNativeColumnarToRowExec])
+         }
+       }
+     }
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
 index fee375db10a..8c2c24e2c5f 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala

From cfa898f9255b467dfc8ceb24316cb931361d25af Mon Sep 17 00:00:00 2001
From: comphead <comphead@ukr.net>
Date: Wed, 29 Jul 2026 08:24:01 -0700
Subject: [PATCH 8/9] address today's review comments

- Restore the removed comment above legacyCastComplexTypesToStringReason
  explaining the legacy formatting differences, updated to describe the
  codegen dispatcher routing instead of a Spark fallback.
- Drop the "natively" qualifier from legacyCastComplexTypesToStringReason.
  Incompatibility reasons already document ways the native implementation
  differs from Spark, so the qualifier is redundant with the docstring on
  getIncompatibleReasons.
- Restore the removed comment on the negative-scale decimal to string test
  in CometCastSuite, explaining why localTableScan is toggled on.
---
 .../org/apache/comet/expressions/CometCast.scala    |  8 +++++++-
 .../scala/org/apache/comet/CometCastSuite.scala     | 13 +++++++++++++
 2 files changed, 20 insertions(+), 1 deletion(-)

diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index b4cad10d6e..7b5b124585 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -41,8 +41,14 @@ object CometCast
   private[comet] val negativeScaleDecimalToStringReason: String =
     "Negative-scale decimal requires spark.sql.legacy.allowNegativeScaleOfDecimal=true"
 
+  // When `spark.sql.legacy.castComplexTypesToString.enabled` is true, Spark wraps maps and
+  // structs with `[]` (instead of `{}`) when casting to string, and omits NULL elements of
+  // structs/maps/arrays (instead of rendering them as the literal "null"). Comet's native
+  // cast only implements the default formatting, so with the flag enabled we route
+  // array/map/struct to-string casts through the JVM codegen dispatcher via
+  // `CodegenDispatchFallback`. The flag is internal in Spark 4.0 and defaults to false.
   private[comet] val legacyCastComplexTypesToStringReason: String =
-    "spark.sql.legacy.castComplexTypesToString.enabled=true is not supported natively"
+    "spark.sql.legacy.castComplexTypesToString.enabled=true is not supported"
 
   private def legacyCastComplexTypesToString: Boolean =
     SQLConf.get
diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
index e0c0299144..b611e5a2ca 100644
--- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
@@ -735,6 +735,19 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("cast DecimalType with negative scale to StringType") {
+    // Negative-scale decimals are a legacy Spark feature gated on
+    // spark.sql.legacy.allowNegativeScaleOfDecimal=true. Spark LEGACY cast uses Java's
+    // BigDecimal.toString() which produces scientific notation for negative-scale values
+    // (e.g. 12300 stored as Decimal(7,-2) with unscaled=123 → "1.23E+4").
+    // CometCast.canCastToString checks the
+    // config and returns Incompatible when it is false.
+    //
+    // Parquet does not support negative-scale decimals so we use checkSparkAnswer directly
+    // (no parquet round-trip) to avoid schema coercion.
+
+    // With config enabled, enable localTableScan so Comet can take over the full plan
+    // and execute the cast natively. Parquet does not support negative-scale decimals so
+    // the data is kept in-memory; localTableScan.enabled bridges that gap.
     withSQLConf(
       "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true",
       "spark.comet.exec.localTableScan.enabled" -> "true") {

From fe641fa0fc29c50a578fd2151ed3924f60e76790 Mon Sep 17 00:00:00 2001
From: comphead <comphead@ukr.net>
Date: Wed, 29 Jul 2026 08:32:09 -0700
Subject: [PATCH 9/9] refactor(cast): dedupe VariantType guard and trim
 redundant prose

Fold the outer `isVariantType` guard in `CometCast.getSupportLevel` into the
`Literal` short-circuit branch. The non-literal path already runs the same
guard via `isSupported`, so keeping both fires it twice per plan. Reorder
`isComplexType(fromType) && legacyCastComplexTypesToString` so the SQLConf
lookup only runs when the source is complex. Trim duplicate justification
comments across `CometCast.scala`, the two `cast_complex_types_to_string*.sql`
headers, and the `VariantType` test in `CometCastSuite`.
---
 .../apache/comet/expressions/CometCast.scala  | 30 ++++++++-----------
 .../cast/cast_complex_types_to_string.sql     |  5 ++--
 .../cast_complex_types_to_string_legacy.sql   | 19 +++++-------
 .../org/apache/comet/CometCastSuite.scala     |  8 ++---
 4 files changed, 25 insertions(+), 37 deletions(-)

diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index 7b5b124585..eba1e4b152 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -78,19 +78,15 @@ object CometCast
   // would only duplicate the matrix and risk drifting from it.
 
   override def getSupportLevel(cast: Cast): SupportLevel = {
-    // Reject `VariantType` before the Literal short-circuit below. Folding a Cast whose child or
-    // target is `VariantType` produces a `Literal[VariantType]` that no downstream Comet serde
-    // can serialize, and relying on `CometLiteral` to reject it after the fact leaves a native
-    // path that assumes the produced literal is safe. Report `Unsupported`; the
-    // `CodegenDispatchFallback` mixin will then try the codegen dispatcher, which itself cannot
-    // serialize `VariantType` data args or return types, so the operator falls back to Spark.
-    if (isVariantType(cast.child.dataType) || isVariantType(cast.dataType)) {
-      return unsupported(cast.child.dataType, cast.dataType)
-    }
     if (cast.child.isInstanceOf[Literal]) {
       // A cast whose child is a literal is folded by Spark at planning time via `cast.eval()`
       // (see `convert`), so the cast never executes natively and the result matches Spark by
-      // definition. `CometLiteral` then validates the resulting literal's data type.
+      // definition. `CometLiteral` then validates the resulting literal's data type, except
+      // for `VariantType` which must be rejected here: the fold produces a `Literal[VariantType]`
+      // that no downstream Comet serde can serialize.
+      if (isVariantType(cast.child.dataType) || isVariantType(cast.dataType)) {
+        return unsupported(cast.child.dataType, cast.dataType)
+      }
       Compatible()
     } else {
       isSupported(cast.child.dataType, cast.dataType, cast.timeZoneId, evalMode(cast))
@@ -174,12 +170,10 @@ object CometCast
       timeZoneId: Option[String],
       evalMode: CometEvalMode.Value): SupportLevel = {
 
-    // Spark 4's `VariantType` (SPARK-45827) has no native counterpart in Comet: serializing it
-    // into the DataFusion plan fails in `serializeDataType`, and the codegen dispatcher used by
-    // the `CodegenDispatchFallback` mixin also cannot serialize `VariantType` in the data args or
-    // return type. Detect it via the version-shimmed `isVariantType` (which returns false on
-    // Spark 3.x where the class does not exist) and report `Unsupported`. The mixin will attempt
-    // the codegen dispatcher and then fall back to Spark when the dispatcher rejects the type.
+    // Spark 4's `VariantType` (SPARK-45827) has no native counterpart in Comet, and the codegen
+    // dispatcher also cannot serialize `VariantType` in the data args or return type. The
+    // version-shimmed `isVariantType` returns false on Spark 3.x. Reporting `Unsupported` lets the
+    // `CodegenDispatchFallback` mixin try the dispatcher and then fall back to Spark cleanly.
     if (isVariantType(fromType) || isVariantType(toType)) {
       return unsupported(fromType, toType)
     }
@@ -188,8 +182,8 @@ object CometCast
       return Compatible()
     }
 
-    if (toType == DataTypes.StringType && legacyCastComplexTypesToString && isComplexType(
-        fromType)) {
+    if (toType == DataTypes.StringType && isComplexType(fromType) &&
+      legacyCastComplexTypesToString) {
       return Unsupported(Some(legacyCastComplexTypesToStringReason))
     }
 
diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
index cf16f9f8f5..c9c1dc8ff6 100644
--- a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
+++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
@@ -276,9 +276,8 @@ SELECT cast(array(map('k', 1)) as string)
 -- ----------------------------------------------------------------------------
 -- Map → string
 -- ----------------------------------------------------------------------------
--- Comet does not implement map-to-string casts natively; the CodegenDispatchFallback
--- mixin on CometCast routes them through Spark's doGenCode inside the Comet pipeline,
--- so results match Spark exactly without a full Spark fallback.
+-- Comet has no native map-to-string cast; `CometCast` mixes in `CodegenDispatchFallback`, so
+-- these stay native via the codegen dispatcher and match Spark exactly.
 -- Note: maps materialized through parquet have nondeterministic entry order, so map column
 -- tests use literal maps directly rather than reading from a parquet table.
 
diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql
index 92010a7c89..fcdb94b352 100644
--- a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql
+++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql
@@ -15,28 +15,25 @@
 -- specific language governing permissions and limitations
 -- under the License.
 
--- When `spark.sql.legacy.castComplexTypesToString.enabled` is true Spark wraps maps and
--- structs with `[...]` (instead of `{...}`) and omits NULL elements of structs/maps/arrays
--- (instead of rendering them as the literal "null"). Comet's native cast does not implement
--- the legacy formatting; the [[CodegenDispatchFallback]] mixin on `CometCast` routes these
--- casts through the JVM codegen dispatcher (Spark's own `doGenCode` inside the Comet kernel)
--- so results match Spark exactly without a Spark fallback.
--- The flag is internal in Spark 4.0 and defaults to false.
+-- When `spark.sql.legacy.castComplexTypesToString.enabled` is true, Spark wraps maps and
+-- structs with `[...]` instead of `{...}` and omits NULL elements. `CometCast` mixes in
+-- `CodegenDispatchFallback`, so these casts stay native via the codegen dispatcher and
+-- match Spark exactly. The flag is internal in Spark 4.0 and defaults to false.
 
 -- Config: spark.sql.legacy.castComplexTypesToString.enabled=true
 
--- Struct → string routed through the codegen dispatcher.
+-- Struct → string.
 query
 SELECT CAST(struct(1, 2, null) AS STRING)
 
--- Array → string routed through the codegen dispatcher.
+-- Array → string.
 query
 SELECT CAST(array(1, 2, null) AS STRING)
 
--- Map → string routed through the codegen dispatcher.
+-- Map → string.
 query
 SELECT CAST(map('a', 1, 'b', null) AS STRING)
 
--- Nested complex types also routed through the codegen dispatcher via the outer type.
+-- Nested complex types via the outer struct.
 query
 SELECT CAST(struct(array(1, null), map('k', null)) AS STRING)
diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
index b611e5a2ca..fa5178cdfd 100644
--- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
@@ -1869,11 +1869,9 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("cast to and from VariantType matches Spark") {
-    // Spark 4's VariantType has no native counterpart in Comet: the CometCast
-    // matrix reports Unsupported, and the codegen dispatcher cannot serialize a
-    // VariantType data arg or output type either, so the operator falls back to
-    // Spark cleanly. Guard on Spark 4.0+ because VariantType and parse_json do
-    // not exist in Spark 3.x.
+    // VariantType has no native path in Comet and the codegen dispatcher cannot serialize it
+    // either, so the operator falls back to Spark. Guarded on Spark 4.0+ (parse_json / VARIANT
+    // are unavailable in 3.x).
     assume(CometSparkSessionExtensions.isSpark40Plus, "VariantType requires Spark 4.0+")
     withTable("variant_cast") {
       sql("CREATE TABLE variant_cast(id INT, s STRING) USING parquet")