[SPARK-28141][SQL] Support special date values

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala

@@ -491,7 +491,7 @@ object CatalogColumnStat extends Logging {
  dataType match {
  case BooleanType => s.toBoolean
  case DateType if version == 1 => DateTimeUtils.fromJavaDate(java.sql.Date.valueOf(s))
- case DateType => DateFormatter().parse(s)
+ case DateType => DateFormatter(ZoneOffset.UTC).parse(s)
  case TimestampType if version == 1 =>
  DateTimeUtils.fromJavaTimestamp(java.sql.Timestamp.valueOf(s))
  case TimestampType => getTimestampFormatter().parse(s)
@@ -516,7 +516,7 @@ object CatalogColumnStat extends Logging {
  */
  def toExternalString(v: Any, colName: String, dataType: DataType): String = {
  val externalValue = dataType match {
- case DateType => DateFormatter().format(v.asInstanceOf[Int])
+ case DateType => DateFormatter(ZoneOffset.UTC).format(v.asInstanceOf[Int])
  case TimestampType => getTimestampFormatter().format(v.asInstanceOf[Long])
  case BooleanType | _: IntegralType | FloatType | DoubleType => v
  case _: DecimalType => v.asInstanceOf[Decimal].toJavaBigDecimal

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityGenerator.scala

@@ -45,7 +45,10 @@ class UnivocityGenerator(
  options.timestampFormat,
  options.zoneId,
  options.locale)
- private val dateFormatter = DateFormatter(options.dateFormat, options.locale)
+ private val dateFormatter = DateFormatter(
+ options.dateFormat,
+ options.zoneId,
+ options.locale)
 
  private def makeConverter(dataType: DataType): ValueConverter = dataType match {
  case DateType =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala

@@ -78,7 +78,10 @@ class UnivocityParser(
  options.timestampFormat,
  options.zoneId,
  options.locale)
- private val dateFormatter = DateFormatter(options.dateFormat, options.locale)
+ private val dateFormatter = DateFormatter(
+ options.dateFormat,
+ options.zoneId,
+ options.locale)
 
  // Retrieve the raw record string.
  private def getCurrentInput: UTF8String = {

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

@@ -106,7 +106,7 @@ object Cast {
  * * Cast.castToTimestamp
  */
  def needsTimeZone(from: DataType, to: DataType): Boolean = (from, to) match {
- case (StringType, TimestampType) => true
+ case (StringType, TimestampType | DateType) => true
  case (DateType, TimestampType) => true
  case (TimestampType, StringType) => true
  case (TimestampType, DateType) => true
@@ -287,7 +287,7 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
  // [[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])
 
- private lazy val dateFormatter = DateFormatter()
+ private lazy val dateFormatter = DateFormatter(zoneId)
  private lazy val timestampFormatter = TimestampFormatter.getFractionFormatter(zoneId)
  private val failOnIntegralTypeOverflow = SQLConf.get.failOnIntegralTypeOverflow
 
@@ -469,7 +469,7 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
  // DateConverter
  private[this] def castToDate(from: DataType): Any => Any = from match {
  case StringType =>
- buildCast[UTF8String](_, s => DateTimeUtils.stringToDate(s).orNull)
+ buildCast[UTF8String](_, s => DateTimeUtils.stringToDate(s, zoneId).orNull)
  case TimestampType =>
  // throw valid precision more than seconds, according to Hive.
  // Timestamp.nanos is in 0 to 999,999,999, no more than a second.
@@ -1056,28 +1056,35 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
 
  private[this] def castToDateCode(
  from: DataType,
- ctx: CodegenContext): CastFunction = from match {
- case StringType =>
- val intOpt = ctx.freshVariable("intOpt", classOf[Option[Integer]])
- (c, evPrim, evNull) => code"""
- scala.Option<Integer> $intOpt =
- org.apache.spark.sql.catalyst.util.DateTimeUtils.stringToDate($c);
- if ($intOpt.isDefined()) {
- $evPrim = ((Integer) $intOpt.get()).intValue();
- } else {
- $evNull = true;
- }
- """
- case TimestampType =>
+ ctx: CodegenContext): CastFunction = {
+ def getZoneId() = {
  val zoneIdClass = classOf[ZoneId]
-val zid = JavaCode.global(
+ JavaCode.global(
  ctx.addReferenceObj("zoneId", zoneId, zoneIdClass.getName),
  zoneIdClass)
- (c, evPrim, evNull) =>
- code"""$evPrim =
- org.apache.spark.sql.catalyst.util.DateTimeUtils.microsToEpochDays($c, $zid);"""
- case _ =>
- (c, evPrim, evNull) => code"$evNull = true;"
+ }
+ from match {
+ case StringType =>
+ val intOpt = ctx.freshVariable("intOpt", classOf[Option[Integer]])
+ val zid = getZoneId()
+ (c, evPrim, evNull) =>
+ code"""
+ scala.Option<Integer> $intOpt =
+ org.apache.spark.sql.catalyst.util.DateTimeUtils.stringToDate($c, $zid);
+ if ($intOpt.isDefined()) {
+ $evPrim = ((Integer) $intOpt.get()).intValue();
+ } else {
+ $evNull = true;
+ }
+ """
+ case TimestampType =>
+ val zid = getZoneId()
+ (c, evPrim, evNull) =>
+ code"""$evPrim =
+ org.apache.spark.sql.catalyst.util.DateTimeUtils.microsToEpochDays($c, $zid);"""
+ case _ =>
+ (c, evPrim, evNull) => code"$evNull = true;"
+ }
  }
 
  private[this] def changePrecision(d: ExprValue, decimalType: DecimalType,

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

@@ -76,9 +76,7 @@ case class CurrentDate(timeZoneId: Option[String] = None)
  override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
  copy(timeZoneId = Option(timeZoneId))
 
- override def eval(input: InternalRow): Any = {
- localDateToDays(LocalDate.now(zoneId))
- }
+ override def eval(input: InternalRow): Any = currentDate(zoneId)
 
  override def prettyName: String = "current_date"
 }

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

@@ -371,7 +371,9 @@ case class Literal (value: Any, dataType: DataType) extends LeafExpression {
  case _ => v + "D"
  }
  case (v: Decimal, t: DecimalType) => v + "BD"
- case (v: Int, DateType) => s"DATE '${DateFormatter().format(v)}'"
+ case (v: Int, DateType) =>
+ val formatter = DateFormatter(DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone))
+ s"DATE '${formatter.format(v)}'"
  case (v: Long, TimestampType) =>
  val formatter = TimestampFormatter.getFractionFormatter(
  DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala

@@ -81,7 +81,10 @@ private[sql] class JacksonGenerator(
  options.timestampFormat,
  options.zoneId,
  options.locale)
- private val dateFormatter = DateFormatter(options.dateFormat, options.locale)
+ private val dateFormatter = DateFormatter(
+ options.dateFormat,
+ options.zoneId,
+ options.locale)
 
  private def makeWriter(dataType: DataType): ValueWriter = dataType match {
  case NullType =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala

@@ -59,7 +59,10 @@ class JacksonParser(
  options.timestampFormat,
  options.zoneId,
  options.locale)
- private val dateFormatter = DateFormatter(options.dateFormat, options.locale)
+ private val dateFormatter = DateFormatter(
+ options.dateFormat,
+ options.zoneId,
+ options.locale)
 
  /**
  * Create a converter which converts the JSON documents held by the `JsonParser`

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -1734,7 +1734,8 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
  }
  try {
  valueType match {
- case "DATE" => toLiteral(stringToDate, DateType)
+ case "DATE" =>
+ toLiteral(stringToDate(_, getZoneId(SQLConf.get.sessionLocalTimeZone)), DateType)
  case "TIMESTAMP" =>
  val zoneId = getZoneId(SQLConf.get.sessionLocalTimeZone)
  toLiteral(stringToTimestamp(_, zoneId), TimestampType)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateFormatter.scala

@@ -17,24 +17,30 @@
 
 package org.apache.spark.sql.catalyst.util
 
-import java.time.LocalDate
+import java.time.{LocalDate, ZoneId}
 import java.util.Locale
 
+import DateTimeUtils.{convertSpecialDate, localDateToDays}
+
 sealed trait DateFormatter extends Serializable {
  def parse(s: String): Int // returns days since epoch
  def format(days: Int): String
 }
 
 class Iso8601DateFormatter(
  pattern: String,
+ zoneId: ZoneId,
  locale: Locale) extends DateFormatter with DateTimeFormatterHelper {
 
  @transient
  private lazy val formatter = getOrCreateFormatter(pattern, locale)
 
  override def parse(s: String): Int = {
- val localDate = LocalDate.parse(s, formatter)
- DateTimeUtils.localDateToDays(localDate)
+ val specialDate = convertSpecialDate(s.trim, zoneId)
+ specialDate.getOrElse {
+ val localDate = LocalDate.parse(s, formatter)
+ localDateToDays(localDate)
+ }
  }
 
  override def format(days: Int): String = {
@@ -46,11 +52,13 @@ object DateFormatter {
  val defaultPattern: String = "uuuu-MM-dd"
  val defaultLocale: Locale = Locale.US
 
- def apply(format: String, locale: Locale): DateFormatter = {
- new Iso8601DateFormatter(format, locale)
+ def apply(format: String, zoneId: ZoneId, locale: Locale): DateFormatter = {
+ new Iso8601DateFormatter(format, zoneId, locale)
  }
 
- def apply(format: String): DateFormatter = apply(format, defaultLocale)
+ def apply(format: String, zoneId: ZoneId): DateFormatter = {
+ apply(format, zoneId, defaultLocale)
+ }
 
- def apply(): DateFormatter = apply(defaultPattern)
+ def apply(zoneId: ZoneId): DateFormatter = apply(defaultPattern, zoneId)
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala

@@ -24,7 +24,6 @@ import java.time.temporal.{ChronoField, ChronoUnit, IsoFields}
 import java.util.{Locale, TimeZone}
 import java.util.concurrent.TimeUnit._
 
-import scala.util.Try
 import scala.util.control.NonFatal
 
 import org.apache.spark.sql.types.Decimal
@@ -379,14 +378,16 @@ object DateTimeUtils {
  * `yyyy-[m]m-[d]d *`
  * `yyyy-[m]m-[d]dT*`
  */
- def stringToDate(s: UTF8String): Option[SQLDate] = {
+ def stringToDate(s: UTF8String, zoneId: ZoneId): Option[SQLDate] = {
  if (s == null) {
  return None
  }
  val segments: Array[Int] = Array[Int](1, 1, 1)
  var i = 0
  var currentSegmentValue = 0
  val bytes = s.trim.getBytes
+ val specialDate = convertSpecialDate(bytes, zoneId)
+ if (specialDate.isDefined) return specialDate
  var j = 0
  while (j < bytes.length && (i < 3 && !(bytes(j) == ' ' || bytes(j) == 'T'))) {
  val b = bytes(j)
@@ -855,6 +856,8 @@ object DateTimeUtils {
 
  def currentTimestamp(): SQLTimestamp = instantToMicros(Instant.now())
 
+ def currentDate(zoneId: ZoneId): SQLDate = localDateToDays(LocalDate.now(zoneId))
+
  private def today(zoneId: ZoneId): ZonedDateTime = {
  Instant.now().atZone(zoneId).`with`(LocalTime.MIDNIGHT)
  }
@@ -915,4 +918,28 @@ object DateTimeUtils {
  None
  }
  }
+
+ /**
+ * Converts notational shorthands that are converted to ordinary dates.
+ * @param input - a trimmed string
+ * @param zoneId - zone identifier used to get the current date.
+ * @return some of days since the epoch if the conversion completed successfully otherwise None.
+ */
+ def convertSpecialDate(input: String, zoneId: ZoneId): Option[SQLDate] = {
+ extractSpecialValue(input, zoneId).flatMap {
+ case "epoch" => Some(0)
+ case "now" | "today" => Some(currentDate(zoneId))
+ case "tomorrow" => Some(Math.addExact(currentDate(zoneId), 1))
+ case "yesterday" => Some(Math.subtractExact(currentDate(zoneId), 1))
+ case _ => None
+ }
+ }
+
+ private def convertSpecialDate(bytes: Array[Byte], zoneId: ZoneId): Option[SQLDate] = {
+ if (bytes.length > 0 && Character.isAlphabetic(bytes(0))) {
+ convertSpecialDate(new String(bytes, StandardCharsets.UTF_8), zoneId)
+ } else {
+ None
+ }
+ }
 }

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

@@ -174,7 +174,7 @@ class HashExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
  test("hive-hash for date type") {
  def checkHiveHashForDateType(dateString: String, expected: Long): Unit = {
  checkHiveHash(
- DateTimeUtils.stringToDate(UTF8String.fromString(dateString)).get,
+ DateTimeUtils.stringToDate(UTF8String.fromString(dateString), ZoneOffset.UTC).get,
  DateType,
  expected)
  }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala

@@ -19,7 +19,7 @@ package org.apache.spark.sql.catalyst.util
 
 import java.sql.{Date, Timestamp}
 import java.text.SimpleDateFormat
-import java.time.{LocalDateTime, LocalTime, ZoneId}
+import java.time.{LocalDate, LocalDateTime, LocalTime, ZoneId, ZoneOffset}
 import java.util.{Locale, TimeZone}
 import java.util.concurrent.TimeUnit
 
@@ -120,28 +120,32 @@ class DateTimeUtilsSuite extends SparkFunSuite with Matchers {
  checkFromToJavaDate(new Date(df2.parse("1776-07-04 18:30:00 UTC").getTime))
  }
 
+ private def toDate(s: String, zoneId: ZoneId = ZoneOffset.UTC): Option[SQLDate] = {
+ stringToDate(UTF8String.fromString(s), zoneId)
+ }
+
  test("string to date") {
- assert(stringToDate(UTF8String.fromString("2015-01-28")).get === days(2015, 1, 28))
- assert(stringToDate(UTF8String.fromString("2015")).get === days(2015, 1, 1))
- assert(stringToDate(UTF8String.fromString("0001")).get === days(1, 1, 1))
- assert(stringToDate(UTF8String.fromString("2015-03")).get === days(2015, 3, 1))
+ assert(toDate("2015-01-28").get === days(2015, 1, 28))
+ assert(toDate("2015").get === days(2015, 1, 1))
+ assert(toDate("0001").get === days(1, 1, 1))
+ assert(toDate("2015-03").get === days(2015, 3, 1))
  Seq("2015-03-18", "2015-03-18 ", " 2015-03-18", " 2015-03-18 ", "2015-03-18 123142",
  "2015-03-18T123123", "2015-03-18T").foreach { s =>
- assert(stringToDate(UTF8String.fromString(s)).get === days(2015, 3, 18))
+ assert(toDate(s).get === days(2015, 3, 18))
  }
 
- assert(stringToDate(UTF8String.fromString("2015-03-18X")).isEmpty)
- assert(stringToDate(UTF8String.fromString("2015/03/18")).isEmpty)
- assert(stringToDate(UTF8String.fromString("2015.03.18")).isEmpty)
- assert(stringToDate(UTF8String.fromString("20150318")).isEmpty)
- assert(stringToDate(UTF8String.fromString("2015-031-8")).isEmpty)
- assert(stringToDate(UTF8String.fromString("02015-03-18")).isEmpty)
- assert(stringToDate(UTF8String.fromString("015-03-18")).isEmpty)
- assert(stringToDate(UTF8String.fromString("015")).isEmpty)
- assert(stringToDate(UTF8String.fromString("02015")).isEmpty)
- assert(stringToDate(UTF8String.fromString("1999 08 01")).isEmpty)
- assert(stringToDate(UTF8String.fromString("1999-08 01")).isEmpty)
- assert(stringToDate(UTF8String.fromString("1999 08")).isEmpty)
+ assert(toDate("2015-03-18X").isEmpty)
+ assert(toDate("2015/03/18").isEmpty)
+ assert(toDate("2015.03.18").isEmpty)
+ assert(toDate("20150318").isEmpty)
+ assert(toDate("2015-031-8").isEmpty)
+ assert(toDate("02015-03-18").isEmpty)
+ assert(toDate("015-03-18").isEmpty)
+ assert(toDate("015").isEmpty)
+ assert(toDate("02015").isEmpty)
+ assert(toDate("1999 08 01").isEmpty)
+ assert(toDate("1999-08 01").isEmpty)
+ assert(toDate("1999 08").isEmpty)
  }
 
  private def toTimestamp(str: String, zoneId: ZoneId): Option[SQLTimestamp] = {
@@ -264,12 +268,10 @@ class DateTimeUtilsSuite extends SparkFunSuite with Matchers {
 
  test("SPARK-15379: special invalid date string") {
  // Test stringToDate
- assert(stringToDate(
- UTF8String.fromString("2015-02-29 00:00:00")).isEmpty)
- assert(stringToDate(
- UTF8String.fromString("2015-04-31 00:00:00")).isEmpty)
- assert(stringToDate(UTF8String.fromString("2015-02-29")).isEmpty)
- assert(stringToDate(UTF8String.fromString("2015-04-31")).isEmpty)
+ assert(toDate("2015-02-29 00:00:00").isEmpty)
+ assert(toDate("2015-04-31 00:00:00").isEmpty)
+ assert(toDate("2015-02-29").isEmpty)
+ assert(toDate("2015-04-31").isEmpty)
 
 
  // Test stringToTimestamp
@@ -586,4 +588,16 @@ class DateTimeUtilsSuite extends SparkFunSuite with Matchers {
  toTimestamp(" tomorrow CET ", zoneId).get should be (today + MICROS_PER_DAY +- tolerance)
  }
  }
+
+ test("special date values") {
+ DateTimeTestUtils.outstandingZoneIds.foreach { zoneId =>
+ assert(toDate("epoch", zoneId).get === 0)
+ val today = localDateToDays(LocalDate.now(zoneId))
+ assert(toDate("YESTERDAY", zoneId).get === today - 1)
+ assert(toDate(" Now ", zoneId).get === today)
+ assert(toDate("now UTC", zoneId) === None) // "now" does not accept time zones
+ assert(toDate("today", zoneId).get === today)
+ assert(toDate("tomorrow CET ", zoneId).get === today + 1)
+ }
+ }
 }