Licht-T commented on code in PR #55736:
URL: https://github.com/apache/spark/pull/55736#discussion_r3453709007
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala:
##########
@@ -486,69 +488,84 @@ object DateTimeUtils extends SparkDateTimeUtils {
}
/**
- * Fast path for truncating to MINUTE/HOUR/DAY using offset arithmetic
instead of
- * allocating a `ZonedDateTime` per row. The offset is resolved once for
`micros`; the
- * truncation then runs as `floorMod` in local time. We fall back to
[[truncToUnit]] when
- * the offset at the candidate truncated instant differs from the offset at
`micros`,
- * which means the truncation crosses a DST/historical transition and the
local-time
- * alignment we computed is no longer valid (see SPARK-30766/30857). The
check is
- * skipped for fixed-offset zones. Sub-minute offsets (e.g.
America/Los_Angeles LMT
- * -07:52:58, see SPARK-33404) and 30/45-minute offsets (Asia/Kolkata
+05:30, Asia/Kathmandu
- * +05:45) are handled correctly by this path because the offset is applied
as part of
- * the arithmetic; no offset-alignment guard is needed.
+ * Returns the trunc date time from original date time and trunc level.
+ * Trunc level should be generated using `parseTruncLevel()`, should be
between 0 and 9.
+ *
+ * Uses an offset-arithmetic fast path: the zone offset at `micros` is
resolved once,
+ * truncation runs in the shifted-local frame, and the result is shifted
back to UTC
+ * micros. Falls back to [[truncTimestampSlow]] when the offset at the
candidate
+ * truncated instant differs from the offset at `micros` (DST/historical
transition
+ * spans the candidate; SPARK-30766/30857) or on arithmetic overflow.
*
- * `unitMicros` must evenly divide `MICROS_PER_DAY`; otherwise wall-clock
unit
- * boundaries do not align to multiples of `unitMicros` in the shifted-local
frame
- * and the `floorMod` truncation is unsafe.
+ * Sub-minute LMT offsets (e.g. America/Los_Angeles -07:52:58 pre-1883, see
+ * SPARK-33404) and 30/45-minute offsets (Asia/Kolkata +05:30,
Asia/Kathmandu +05:45)
+ * are handled correctly because the offset is applied as part of the
arithmetic; no
+ * offset-alignment guard is needed.
+ *
+ * The local-frame truncation is selected by `level`:
+ * - MICROSECOND/MILLISECOND/SECOND: no zone information needed (zone
offsets have
+ * at most second precision, see `java.time.ZoneOffset`); reduces to
pure UTC
+ * `floorMod`.
+ * - MINUTE/HOUR/DAY: `floorMod` against the corresponding `unitMicros`.
The unit
+ * must evenly divide `MICROS_PER_DAY`, which holds for all three;
otherwise
+ * wall-clock unit boundaries would not align to multiples of
`unitMicros` in
+ * the shifted-local frame.
+ * - WEEK/MONTH/QUARTER/YEAR: convert local micros to local epoch-day, run
+ * [[truncDate]] in the local-day frame, multiply back to local micros.
*/
- private def truncToUnitFast(
- micros: Long, zoneId: ZoneId, unitMicros: Long, fallbackUnit:
ChronoUnit): Long = {
+ def truncTimestamp(micros: Long, level: Int, zoneId: ZoneId): Long = {
+ // MICROSECOND / MILLISECOND / SECOND don't need zone information.
+ level match {
+ case TRUNC_TO_MICROSECOND => return micros
+ case TRUNC_TO_MILLISECOND => return micros - Math.floorMod(micros,
MICROS_PER_MILLIS)
+ case TRUNC_TO_SECOND => return micros - Math.floorMod(micros,
MICROS_PER_SECOND)
+ case _ =>
+ }
val rules = zoneId.getRules
val originalSec = Math.floorDiv(micros, MICROS_PER_SECOND)
val originalOffsetSec =
rules.getOffset(Instant.ofEpochSecond(originalSec)).getTotalSeconds.toLong
val offsetMicros = originalOffsetSec * MICROS_PER_SECOND
try {
val local = Math.addExact(micros, offsetMicros)
- val truncatedLocal = local - Math.floorMod(local, unitMicros)
+ val truncatedLocal = level match {
+ case TRUNC_TO_MINUTE => local - Math.floorMod(local, MICROS_PER_MINUTE)
+ case TRUNC_TO_HOUR => local - Math.floorMod(local, MICROS_PER_HOUR)
+ case TRUNC_TO_DAY => local - Math.floorMod(local, MICROS_PER_DAY)
+ case _ => // Date-level truncation: WEEK / MONTH / QUARTER / YEAR.
+ val localDays = Math.floorDiv(local, MICROS_PER_DAY).toInt
+ truncDate(localDays, level).toLong * MICROS_PER_DAY
+ }
val candidate = Math.subtractExact(truncatedLocal, offsetMicros)
if (!rules.isFixedOffset) {
val candidateSec = Math.floorDiv(candidate, MICROS_PER_SECOND)
val candidateOffsetSec =
rules.getOffset(Instant.ofEpochSecond(candidateSec)).getTotalSeconds.toLong
if (candidateOffsetSec != originalOffsetSec) {
- return truncToUnit(micros, zoneId, fallbackUnit)
+ return truncTimestampSlow(micros, level, zoneId)
}
}
candidate
} catch {
- case _: ArithmeticException => truncToUnit(micros, zoneId, fallbackUnit)
+ case _: ArithmeticException => truncTimestampSlow(micros, level, zoneId)
}
}
/**
- * Returns the trunc date time from original date time and trunc level.
- * Trunc level should be generated using `parseTruncLevel()`, should be
between 0 and 9.
+ * Slow-path reference implementation: equivalent to [[truncTimestamp]]
without the
+ * offset-arithmetic fast path. Invoked from [[truncTimestamp]] on DST-cross
or
Review Comment:
Thanks, @MaxGekk. Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
