refactor cast

Author: andygrove

spark/src/main/scala/org/apache/comet/GenerateDocs.scala

@@ -25,7 +25,8 @@ import scala.collection.mutable.ListBuffer
 
 import org.apache.spark.sql.catalyst.expressions.Cast
 
-import org.apache.comet.expressions.{CometCast, CometEvalMode, Compatible, Incompatible}
+import org.apache.comet.expressions.CometEvalMode
+import org.apache.comet.serde.{CometCast, Compatible, Incompatible}
 
 /**
  * Utility for generating markdown documentation from the configs.

spark/src/main/scala/org/apache/comet/expressions/CometCast.scala renamed to spark/src/main/scala/org/apache/comet/serde/CometCast.scala

@@ -17,9 +17,14 @@
  * under the License.
  */
 
-package org.apache.comet.expressions
+package org.apache.comet.serde
 
-import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DecimalType, NullType, StructType}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast}
+import org.apache.spark.sql.types._
+
+import org.apache.comet.expressions.CometEvalMode
+import org.apache.comet.serde.QueryPlanSerde.handleCast
+import org.apache.comet.shims.CometExprShim
 
 sealed trait SupportLevel
 
@@ -32,7 +37,20 @@ case class Incompatible(notes: Option[String] = None) extends SupportLevel
 /** We do not support this feature */
 object Unsupported extends SupportLevel
 
-object CometCast {
+object CometCast
+    extends CometExpressionSerde[Cast]
+    with CometExprShim
+    with IncompatExpr
+    with IncompatAnsiExpr[Cast] {
+
+  override def isAnsiMode(expr: Cast): Boolean = expr.ansiEnabled
+
+  override def convert(
+      expr: Cast,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[ExprOuterClass.Expr] = {
+    handleCast(expr, expr.child, inputs, binding, expr.dataType, expr.timeZoneId, evalMode(expr))
+  }
 
   def supportedTypes: Seq[DataType] =
     Seq(

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -169,7 +169,8 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[DateAdd] -> CometDateAdd,
     classOf[DateSub] -> CometDateSub,
     classOf[TruncDate] -> CometTruncDate,
-    classOf[TruncTimestamp] -> CometTruncTimestamp)
+    classOf[TruncTimestamp] -> CometTruncTimestamp,
+    classOf[Cast] -> CometCast)
 
   /**
    * Mapping of Spark aggregate expression class to Comet expression handler.
@@ -689,9 +690,6 @@ object QueryPlanSerde extends Logging with CometExprShim {
           Some(timeZoneId),
           CometEvalMode.TRY)
 
-      case c @ Cast(child, dt, timeZoneId, _) =>
-        handleCast(expr, child, inputs, binding, dt, timeZoneId, evalMode(c))
-
       case EqualTo(left, right) =>
         createBinaryExpr(
           expr,

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

@@ -33,7 +33,8 @@ import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{DataType, DataTypes, DecimalType, StructField, StructType}
 
 import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus
-import org.apache.comet.expressions.{CometCast, CometEvalMode, Compatible}
+import org.apache.comet.expressions.CometEvalMode
+import org.apache.comet.serde.{CometCast, Compatible}
 
 class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
 

spark/src/test/spark-3.5/org/apache/spark/sql/CometToPrettyStringSuite.scala

@@ -20,7 +20,8 @@
 package org.apache.spark.sql
 
 import org.apache.comet.CometConf
-import org.apache.comet.expressions.{CometCast, CometEvalMode, Compatible}
+import org.apache.comet.expressions.CometEvalMode
+import org.apache.comet.serde.{CometCast, Compatible}
 import org.apache.comet.testing.{DataGenOptions, ParquetGenerator}
 import org.apache.commons.io.FileUtils
 import org.apache.spark.sql.catalyst.TableIdentifier