feat: [branch-1.0] add codegen dispatch fallback for CometCast incompatible/unsupported cases (#5079)

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

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

spark/src/main/scala/org/apache/comet/expressions/CometCast.scala

@@ -25,22 +25,28 @@ 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.
+  // 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"
 
@@ -75,7 +81,12 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
     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))
@@ -159,13 +170,20 @@ 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, 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)
+    }
+
     if (fromType == toType) {
       return Compatible()
     }
 
-    if (toType == DataTypes.StringType && legacyCastComplexTypesToString && (fromType
-        .isInstanceOf[ArrayType] || fromType.isInstanceOf[StructType] ||
-        fromType.isInstanceOf[MapType])) {
+    if (toType == DataTypes.StringType && isComplexType(fromType) &&
+      legacyCastComplexTypesToString) {
       return Unsupported(Some(legacyCastComplexTypesToStringReason))
     }
 

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
 }

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")
 

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,69 +269,70 @@ 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)
+-- 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 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.
 
 -- 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 +341,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)

spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql

@@ -15,26 +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 only implements the default
--- formatting, so any array/map/struct → string cast must fall back to Spark.
--- 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 falls back.
-query expect_fallback(spark.sql.legacy.castComplexTypesToString.enabled=true is not supported)
+-- Struct → string.
+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.
+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.
+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 via the outer struct.
+query
 SELECT CAST(struct(array(1, null), map('k', null)) AS STRING)

spark/src/test/scala/org/apache/comet/CometCastSuite.scala

@@ -1797,7 +1797,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))
@@ -1814,9 +1814,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")))
         }
       }
     }
@@ -1870,6 +1868,26 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("cast to and from VariantType matches Spark") {
+    // 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")
+      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,

spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala

@@ -2060,9 +2060,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))"
@@ -2080,9 +2077,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