ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-36590][SQL] Convert special timestamp_ntz values in the session time zone

Browse source 
In the PR, I propose to use the session time zone ( see the SQL config `spark.sql.session.timeZone`) instead of JVM default time zone while converting of special timestamp_ntz strings such as "today", "tomorrow" and so on.

Current implementation is based on the system time zone, and it controverses to other functions/classes that use the session time zone. For example, Spark doesn't respects user's settings:
```sql
$ export TZ="Europe/Amsterdam"
$ ./bin/spark-sql -S
spark-sql> select timestamp_ntz'now';
2021-08-25 18:12:36.233

spark-sql> set spark.sql.session.timeZone=America/Los_Angeles;
spark.sql.session.timeZone	America/Los_Angeles
spark-sql> select timestamp_ntz'now';
2021-08-25 18:14:40.547
```

Yes. For the example above, after the changes:
```sql
spark-sql> select timestamp_ntz'now';
2021-08-25 18:47:46.832

spark-sql> set spark.sql.session.timeZone=America/Los_Angeles;
spark.sql.session.timeZone	America/Los_Angeles
spark-sql> select timestamp_ntz'now';
2021-08-25 09:48:05.211
```

By running the affected test suites:
```
$ build/sbt "test:testOnly *DateTimeUtilsSuite"
```

Closes #33838 from MaxGekk/fix-ts_ntz-special-values.

Authored-by: Max Gekk <max.gekk@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
(cherry picked from commit 159ff9fd14)
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
This commit is contained in:
Max Gekk 2021-08-26 10:09:18 +08:00 committed by Wenchen Fan
parent 5198c0c316
commit 0c364e607d
5 changed files with 38 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala
View file

@ -129,8 +129,7 @@ object SpecialDatetimeValues extends Rule[LogicalPlan] {
private val conv = Map[DataType, (String, java.time.ZoneId) => Option[Any]]( private val conv = Map[DataType, (String, java.time.ZoneId) => Option[Any]](
DateType -> convertSpecialDate, DateType -> convertSpecialDate,
TimestampType -> convertSpecialTimestamp, TimestampType -> convertSpecialTimestamp,
TimestampNTZType -> ((s: String, _: java.time.ZoneId) => convertSpecialTimestampNTZ(s)) TimestampNTZType -> convertSpecialTimestampNTZ)
)
def apply(plan: LogicalPlan): LogicalPlan = { def apply(plan: LogicalPlan): LogicalPlan = {
plan.transformAllExpressionsWithPruning(_.containsPattern(CAST)) { plan.transformAllExpressionsWithPruning(_.containsPattern(CAST)) {
case cast @ Cast(e, dt @ (DateType | TimestampType | TimestampNTZType), _, _) case cast @ Cast(e, dt @ (DateType | TimestampType | TimestampNTZType), _, _)

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

@ -2134,25 +2134,27 @@ class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with SQLConfHelper with Logg
specialDate.getOrElse(toLiteral(stringToDate, DateType)) specialDate.getOrElse(toLiteral(stringToDate, DateType))
// SPARK-36227: Remove TimestampNTZ type support in Spark 3.2 with minimal code changes. // SPARK-36227: Remove TimestampNTZ type support in Spark 3.2 with minimal code changes.
case "TIMESTAMP_NTZ" if isTesting => case "TIMESTAMP_NTZ" if isTesting =>
val specialTs = convertSpecialTimestampNTZ(value).map(Literal(_, TimestampNTZType)) convertSpecialTimestampNTZ(value, getZoneId(conf.sessionLocalTimeZone))
specialTs.getOrElse(toLiteral(stringToTimestampWithoutTimeZone, TimestampNTZType)) .map(Literal(_, TimestampNTZType))
.getOrElse(toLiteral(stringToTimestampWithoutTimeZone, TimestampNTZType))
case "TIMESTAMP_LTZ" if isTesting => case "TIMESTAMP_LTZ" if isTesting =>
constructTimestampLTZLiteral(value) constructTimestampLTZLiteral(value)
case "TIMESTAMP" => case "TIMESTAMP" =>
SQLConf.get.timestampType match { SQLConf.get.timestampType match {
case TimestampNTZType => case TimestampNTZType =>
val specialTs = convertSpecialTimestampNTZ(value).map(Literal(_, TimestampNTZType)) convertSpecialTimestampNTZ(value, getZoneId(conf.sessionLocalTimeZone))
specialTs.getOrElse { .map(Literal(_, TimestampNTZType))
val containsTimeZonePart = .getOrElse {
DateTimeUtils.parseTimestampString(UTF8String.fromString(value))._2.isDefined val containsTimeZonePart =
// If the input string contains time zone part, return a timestamp with local time DateTimeUtils.parseTimestampString(UTF8String.fromString(value))._2.isDefined
// zone literal. // If the input string contains time zone part, return a timestamp with local time
if (containsTimeZonePart) { // zone literal.
constructTimestampLTZLiteral(value) if (containsTimeZonePart) {
} else { constructTimestampLTZLiteral(value)
toLiteral(stringToTimestampWithoutTimeZone, TimestampNTZType) } else {
toLiteral(stringToTimestampWithoutTimeZone, TimestampNTZType)
}
} }
}
case TimestampType => case TimestampType =>
constructTimestampLTZLiteral(value) constructTimestampLTZLiteral(value)

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

@ -1043,7 +1043,7 @@ object DateTimeUtils {
* Converts notational shorthands that are converted to ordinary timestamps. * Converts notational shorthands that are converted to ordinary timestamps.
* *
* @param input A string to parse. It can contain trailing or leading whitespaces. * @param input A string to parse. It can contain trailing or leading whitespaces.
* @param zoneId Zone identifier used to get the current date. * @param zoneId Zone identifier used to get the current timestamp.
* @return Some of microseconds since the epoch if the conversion completed * @return Some of microseconds since the epoch if the conversion completed
* successfully otherwise None. * successfully otherwise None.
*/ */
@ -1063,18 +1063,19 @@ object DateTimeUtils {
* Converts notational shorthands that are converted to ordinary timestamps without time zone. * Converts notational shorthands that are converted to ordinary timestamps without time zone.
* *
* @param input A string to parse. It can contain trailing or leading whitespaces. * @param input A string to parse. It can contain trailing or leading whitespaces.
* @param zoneId Zone identifier used to get the current local timestamp.
* @return Some of microseconds since the epoch if the conversion completed * @return Some of microseconds since the epoch if the conversion completed
* successfully otherwise None. * successfully otherwise None.
*/ */
def convertSpecialTimestampNTZ(input: String): Option[Long] = { def convertSpecialTimestampNTZ(input: String, zoneId: ZoneId): Option[Long] = {
val localDateTime = extractSpecialValue(input.trim).flatMap { val localDateTime = extractSpecialValue(input.trim).flatMap {
case "epoch" => Some(LocalDateTime.of(1970, 1, 1, 0, 0)) case "epoch" => Some(LocalDateTime.of(1970, 1, 1, 0, 0))
case "now" => Some(LocalDateTime.now()) case "now" => Some(LocalDateTime.now(zoneId))
case "today" => Some(LocalDateTime.now().`with`(LocalTime.MIDNIGHT)) case "today" => Some(LocalDateTime.now(zoneId).`with`(LocalTime.MIDNIGHT))
case "tomorrow" => case "tomorrow" =>
Some(LocalDateTime.now().`with`(LocalTime.MIDNIGHT).plusDays(1)) Some(LocalDateTime.now(zoneId).`with`(LocalTime.MIDNIGHT).plusDays(1))
case "yesterday" => case "yesterday" =>
Some(LocalDateTime.now().`with`(LocalTime.MIDNIGHT).minusDays(1)) Some(LocalDateTime.now(zoneId).`with`(LocalTime.MIDNIGHT).minusDays(1))
case _ => None case _ => None
} }
localDateTime.map(localDateTimeToMicros) localDateTime.map(localDateTimeToMicros)

6
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/SpecialDatetimeValuesSuite.scala
View file

@ -91,9 +91,9 @@ class SpecialDatetimeValuesSuite extends PlanTest {
testSpecialDatetimeValues { zoneId => testSpecialDatetimeValues { zoneId =>
val expected = Set( val expected = Set(
LocalDateTime.of(1970, 1, 1, 0, 0), LocalDateTime.of(1970, 1, 1, 0, 0),
LocalDateTime.now(), LocalDateTime.now(zoneId),
LocalDateTime.now().`with`(LocalTime.MIDNIGHT).plusDays(1), LocalDateTime.now(zoneId).`with`(LocalTime.MIDNIGHT).plusDays(1),
LocalDateTime.now().`with`(LocalTime.MIDNIGHT).minusDays(1) LocalDateTime.now(zoneId).`with`(LocalTime.MIDNIGHT).minusDays(1)
).map(localDateTimeToMicros) ).map(localDateTimeToMicros)
testSpecialTs(TimestampNTZType, expected, zoneId) testSpecialTs(TimestampNTZType, expected, zoneId)
} }

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

@ -793,16 +793,18 @@ class DateTimeUtilsSuite extends SparkFunSuite with Matchers with SQLHelper {
test("SPARK-35979: special timestamp without time zone values") { test("SPARK-35979: special timestamp without time zone values") {
val tolerance = TimeUnit.SECONDS.toMicros(30) val tolerance = TimeUnit.SECONDS.toMicros(30)
assert(convertSpecialTimestampNTZ("Epoch").get === 0) testSpecialDatetimeValues { zoneId =>
val now = DateTimeUtils.localDateTimeToMicros(LocalDateTime.now()) assert(convertSpecialTimestampNTZ("Epoch", zoneId).get === 0)
convertSpecialTimestampNTZ("NOW").get should be(now +- tolerance) val now = DateTimeUtils.localDateTimeToMicros(LocalDateTime.now(zoneId))
val localToday = LocalDateTime.now().`with`(LocalTime.MIDNIGHT) convertSpecialTimestampNTZ("NOW", zoneId).get should be(now +- tolerance)
val yesterday = DateTimeUtils.localDateTimeToMicros(localToday.minusDays(1)) val localToday = LocalDateTime.now(zoneId).`with`(LocalTime.MIDNIGHT)
convertSpecialTimestampNTZ(" Yesterday").get should be(yesterday) val yesterday = DateTimeUtils.localDateTimeToMicros(localToday.minusDays(1))
val today = DateTimeUtils.localDateTimeToMicros(localToday) convertSpecialTimestampNTZ(" Yesterday", zoneId).get should be(yesterday)
convertSpecialTimestampNTZ("Today ").get should be(today) val today = DateTimeUtils.localDateTimeToMicros(localToday)
val tomorrow = DateTimeUtils.localDateTimeToMicros(localToday.plusDays(1)) convertSpecialTimestampNTZ("Today ", zoneId).get should be(today)
convertSpecialTimestampNTZ(" tomorrow ").get should be(tomorrow) val tomorrow = DateTimeUtils.localDateTimeToMicros(localToday.plusDays(1))
convertSpecialTimestampNTZ(" tomorrow ", zoneId).get should be(tomorrow)
}
} }
test("SPARK-28141: special date values") { test("SPARK-28141: special date values") {