[SPARK-29074][SQL] Optimize date_format for foldable fmt

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

@@ -580,7 +580,11 @@ case class WeekOfYear(child: Expression) extends UnaryExpression with ImplicitCa
   """,
   since = "1.5.0")
 // scalastyle:on line.size.limit
-case class DateFormatClass(left: Expression, right: Expression, timeZoneId: Option[String] = None)
+case class DateFormatClass(
+    left: Expression,
+    right: Expression,
+    timeZoneId: Option[String] = None,
+    formatter: Option[TimestampFormatter] = None)
   extends BinaryExpression with TimeZoneAwareExpression with ImplicitCastInputTypes {
 
   def this(left: Expression, right: Expression) = this(left, right, None)
@@ -589,22 +593,41 @@ case class DateFormatClass(left: Expression, right: Expression, timeZoneId: Opti
 
   override def inputTypes: Seq[AbstractDataType] = Seq(TimestampType, StringType)
 
-  override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
-    copy(timeZoneId = Option(timeZoneId))
+  override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression = {
+    val tf = if (formatter.isEmpty && right.foldable) {
+      val format = right.eval().toString
+      Some(TimestampFormatter(
+        format,
+        DateTimeUtils.getZoneId(timeZoneId),
+        Locale.US))
+    } else None
+    copy(formatter = tf, timeZoneId = Option(timeZoneId))
+  }
 
   override protected def nullSafeEval(timestamp: Any, format: Any): Any = {
-    val df = TimestampFormatter(format.toString, zoneId)
-    UTF8String.fromString(df.format(timestamp.asInstanceOf[Long]))
+    val tf = if (formatter.isEmpty) {
+      TimestampFormatter(format.toString, zoneId, Locale.US)
+    } else {
+      formatter.get
+    }
+    UTF8String.fromString(tf.format(timestamp.asInstanceOf[Long]))
   }
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
-    val tf = TimestampFormatter.getClass.getName.stripSuffix("$")
-    val zid = ctx.addReferenceObj("zoneId", zoneId, classOf[ZoneId].getName)
-    val locale = ctx.addReferenceObj("locale", Locale.US)
-    defineCodeGen(ctx, ev, (timestamp, format) => {
-      s"""UTF8String.fromString($tf$$.MODULE$$.apply($format.toString(), $zid, $locale)
+    formatter.map { tf =>
+      val timestampFormatter = ctx.addReferenceObj("timestampFormatter", tf)
+      defineCodeGen(ctx, ev, (timestamp, _) => {
+        s"""UTF8String.fromString($timestampFormatter.format($timestamp))"""
+      })
+    }.getOrElse {
+      val tf = TimestampFormatter.getClass.getName.stripSuffix("$")
+      val zid = ctx.addReferenceObj("zoneId", zoneId, classOf[ZoneId].getName)
+      val locale = ctx.addReferenceObj("locale", Locale.US)
+      defineCodeGen(ctx, ev, (timestamp, format) => {
+        s"""UTF8String.fromString($tf$$.MODULE$$.apply($format.toString(), $zid, $locale)
           .format($timestamp))"""
-    })
+      })
+    }
   }
 
   override def prettyName: String = "date_format"

sql/core/benchmarks/DateTimeBenchmark-results.txt

@@ -168,8 +168,8 @@ Java HotSpot(TM) 64-Bit Server VM 1.8.0_202-b08 on Mac OS X 10.14.3
 Intel(R) Core(TM) i7-4850HQ CPU @ 2.30GHz
 format date:                             Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------
-format date wholestage off                    7180 / 7181          1.4         718.0       1.0X
-format date wholestage on                     7051 / 7194          1.4         705.1       1.0X
+format date wholestage off                    4787 / 4839          2.1         478.7       1.0X
+format date wholestage on                     4736 / 4802          2.1         473.6       1.0X
 
 
 ================================================================================================