Change the implementation to add ignorTimeZone in Canonicalize

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Canonicalize.scala

@@ -27,6 +27,7 @@ package org.apache.spark.sql.catalyst.expressions
  * The following rules are applied:
  *  - Names and nullability hints for [[org.apache.spark.sql.types.DataType]]s are stripped.
  *  - Names for [[GetStructField]] are stripped.
+ *  - TimeZoneId for [[Cast]] and [[AnsiCast]] are stripped if `needsTimeZone` is false.
  *  - Commutative and associative operations ([[Add]] and [[Multiply]]) have their children ordered
  *    by `hashCode`.
  *  - [[EqualTo]] and [[EqualNullSafe]] are reordered by `hashCode`.
@@ -35,7 +36,7 @@ package org.apache.spark.sql.catalyst.expressions
  */
 object Canonicalize {
   def execute(e: Expression): Expression = {
-    expressionReorder(ignoreNamesTypes(e))
+    expressionReorder(ignoreTimeZone(ignoreNamesTypes(e)))
   }
 
   /** Remove names and nullability from types, and names from `GetStructField`. */
@@ -46,6 +47,15 @@ object Canonicalize {
     case _ => e
   }
 
+  /** Remove TimeZoneId for Cast if needsTimeZone return false. */
+  private[expressions] def ignoreTimeZone(e: Expression): Expression = e match {
+    case a: AnsiCast if !a.needsTimeZone && a.timeZoneId.nonEmpty =>
+      a.copy(timeZoneId = None)
+    case c: Cast if !c.needsTimeZone && c.timeZoneId.nonEmpty =>
+      c.copy(timeZoneId = None)
+    case _ => e
+  }
+
   /** Collects adjacent commutative operations. */
   private def gatherCommutative(
       e: Expression,

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -279,7 +279,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
   override lazy val resolved: Boolean =
     childrenResolved && checkInputDataTypes().isSuccess && (!needsTimeZone || timeZoneId.isDefined)
 
-  protected[this] def needsTimeZone: Boolean = Cast.needsTimeZone(child.dataType, dataType)
+  def needsTimeZone: Boolean = Cast.needsTimeZone(child.dataType, dataType)
 
   // [[func]] assumes the input is no longer null because eval already does the null check.
   @inline private[this] def buildCast[T](a: Any, func: T => Any): Any = func(a.asInstanceOf[T])
@@ -1708,11 +1708,6 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
   """)
 case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String] = None)
   extends CastBase {
-  override lazy val canonicalized: Expression = if (!needsTimeZone && timeZoneId.nonEmpty) {
-    copy(timeZoneId = None)
-  } else {
-    this
-  }
 
   override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
     copy(timeZoneId = Option(timeZoneId))
@@ -1730,11 +1725,6 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
  */
 case class AnsiCast(child: Expression, dataType: DataType, timeZoneId: Option[String] = None)
   extends CastBase {
-  override lazy val canonicalized: Expression = if (!needsTimeZone && timeZoneId.nonEmpty) {
-    copy(timeZoneId = None)
-  } else {
-    this
-  }
 
   override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
     copy(timeZoneId = Option(timeZoneId))

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CanonicalizeSuite.scala

@@ -93,7 +93,6 @@ class CanonicalizeSuite extends SparkFunSuite {
     val literal = Literal(1)
     val cast = Cast(literal, LongType)
     val castWithTimeZoneId = Cast(literal, LongType, Some(TimeZone.getDefault.getID))
-
     assert(castWithTimeZoneId.semanticEquals(cast))
   }
 }