Skip to content

[SPARK-31630][SQL] Fix perf regression by skipping timestamps rebasing after some threshold #28441

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
MaxGekk wants to merge 14 commits into from
Closed

[SPARK-31630][SQL] Fix perf regression by skipping timestamps rebasing after some threshold #28441

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
6c536d4
Optimize micros rebasing
MaxGekk May 2, 2020
832f366
Re-gen benchmarks on JDK 8
MaxGekk May 3, 2020
f4187d2
Re-gen results on JDK 11
MaxGekk May 3, 2020
0d00a2b
Merge remote-tracking branch 'origin/master' into opt-rebase-common-t…
MaxGekk May 4, 2020
f8ab34b
Re-gen on JDK 8
MaxGekk May 4, 2020
b9dfb16
Re-gen on JDK 11
MaxGekk May 4, 2020
c6ebcf2
Merge remote-tracking branch 'remotes/origin/master' into opt-rebase-…
MaxGekk May 4, 2020
ded04e2
Make lastSwitchJulianTs public
MaxGekk May 4, 2020
0c44097
Re-gen DateTimeRebaseBenchmark-jdk11-results.txt
MaxGekk May 4, 2020
d31ec0f
Re-gen DateTimeRebaseBenchmark-results.txt
MaxGekk May 4, 2020
dbdc926
Merge remote-tracking branch 'remotes/origin/master' into opt-rebase-…
MaxGekk May 5, 2020
eaca6a8
Improve error message in getLastSwitchTs()
MaxGekk May 5, 2020
7ef27f8
Re-gen DateTimeRebaseBenchmark-jdk11-results.txt
MaxGekk May 5, 2020
2badf88
Re-gen DateTimeRebaseBenchmark-results.txt
MaxGekk May 5, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 34 additions & 13 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,6 +286,17 @@ object RebaseDateTime {
*/
private val gregJulianRebaseMap = loadRebaseRecords("gregorian-julian-rebase-micros.json")

private def getLastSwitchTs(rebaseMap: AnyRefMap[String, RebaseInfo]): Long = {
val latestTs = rebaseMap.values.map(_.switches.last).max
require(rebaseMap.values.forall(_.diffs.last == 0),
Copy link
Contributor

@cloud-fan cloud-fan May 5, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ideally require should be the first line in a method.

Copy link
Member Author

@MaxGekk MaxGekk May 5, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It uses latestTs in the error message. I am going to improve the message by converting micros to Instant, so, toString should form nicer string:

    require(rebaseMap.values.forall(_.diffs.last == 0),
      s"Differences between Julian and Gregorian calendar after ${microsToInstant(latestTs)} " +
      "are expected to be zero for available time zones.")

s"Differences between Julian and Gregorian calendar after ${microsToInstant(latestTs)} " +
"are expected to be zero for all available time zones.")
latestTs
}
// The switch time point after which all diffs between Gregorian and Julian calendars
// across all time zones are zero
private final val lastSwitchGregorianTs: Long = getLastSwitchTs(gregJulianRebaseMap)

private final val gregorianStartTs = LocalDateTime.of(gregorianStartDate, LocalTime.MIDNIGHT)
private final val julianEndTs = LocalDateTime.of(
julianEndDate,
Expand Down Expand Up @@ -344,13 +355,17 @@ object RebaseDateTime {
* @return The rebased microseconds since the epoch in Julian calendar.
*/
def rebaseGregorianToJulianMicros(micros: Long): Long = {
val timeZone = TimeZone.getDefault
val tzId = timeZone.getID
val rebaseRecord = gregJulianRebaseMap.getOrNull(tzId)
if (rebaseRecord == null || micros < rebaseRecord.switches(0)) {
rebaseGregorianToJulianMicros(timeZone.toZoneId, micros)
if (micros >= lastSwitchGregorianTs) {
micros
} else {
rebaseMicros(rebaseRecord, micros)
val timeZone = TimeZone.getDefault
val tzId = timeZone.getID
val rebaseRecord = gregJulianRebaseMap.getOrNull(tzId)
if (rebaseRecord == null || micros < rebaseRecord.switches(0)) {
rebaseGregorianToJulianMicros(timeZone.toZoneId, micros)
} else {
rebaseMicros(rebaseRecord, micros)
}
}
}

Expand Down Expand Up @@ -418,7 +433,9 @@ object RebaseDateTime {
// in the interval: [julianGregDiffSwitchMicros(i), julianGregDiffSwitchMicros(i+1))
private val julianGregRebaseMap = loadRebaseRecords("julian-gregorian-rebase-micros.json")

final val lastSwitchJulianTs: Long = julianGregRebaseMap.values.map(_.switches.last).max
// The switch time point after which all diffs between Julian and Gregorian calendars
// across all time zones are zero
final val lastSwitchJulianTs: Long = getLastSwitchTs(julianGregRebaseMap)

/**
* An optimized version of [[rebaseJulianToGregorianMicros(ZoneId, Long)]]. This method leverages
Expand All @@ -434,13 +451,17 @@ object RebaseDateTime {
* @return The rebased microseconds since the epoch in Proleptic Gregorian calendar.
*/
def rebaseJulianToGregorianMicros(micros: Long): Long = {
val timeZone = TimeZone.getDefault
val tzId = timeZone.getID
val rebaseRecord = julianGregRebaseMap.getOrNull(tzId)
if (rebaseRecord == null || micros < rebaseRecord.switches(0)) {
rebaseJulianToGregorianMicros(timeZone.toZoneId, micros)
if (micros >= lastSwitchJulianTs) {
micros
} else {
rebaseMicros(rebaseRecord, micros)
val timeZone = TimeZone.getDefault
val tzId = timeZone.getID
val rebaseRecord = julianGregRebaseMap.getOrNull(tzId)
if (rebaseRecord == null || micros < rebaseRecord.switches(0)) {
rebaseJulianToGregorianMicros(timeZone.toZoneId, micros)
} else {
rebaseMicros(rebaseRecord, micros)
}
}
}
}
Loading