vranes commented on code in PR #57401:
URL: https://github.com/apache/spark/pull/57401#discussion_r3634478824
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala:
##########
@@ -1266,47 +1266,78 @@ object DateTimeUtils extends SparkDateTimeUtils {
}
/**
- * DayTimeInterval bucketing: bucket k starts at
- * `timestampAddDayTime(originMicros, k * bucketMicros, zoneId)`, matching
the instant that
- * `originMicros + INTERVAL '<k * bucketSize>'` would produce. For sub-day
buckets the
- * calendar-day component is zero, so the result is pure UTC-microsecond
floor division
- * and `zoneId` has no effect.
+ * Start boundary of DayTimeInterval bucket index `k`, on the fixed grid
anchored at
+ * `originMicros`. Sub-day buckets use pure UTC arithmetic; multi-day
buckets add the calendar-day
+ * part in `zoneId`.
*
* `bucketMicros` must be positive; `TimeBucket.checkInputDataTypes`
enforces this at
* analysis time.
*
* @param bucketMicros bucket size in microseconds.
- * @param tsMicros timestamp to bucket, in microseconds since the epoch
(UTC).
+ * @param k bucket index (may be negative).
* @param originMicros grid alignment anchor, in microseconds since the
epoch (UTC).
* @param zoneId zone in which calendar-day arithmetic is performed.
*/
- def timeBucketDTInterval(
- bucketMicros: Long, tsMicros: Long, originMicros: Long, zoneId: ZoneId):
Long = {
- val bucketDays = bucketMicros / MICROS_PER_DAY
+ def timeBucketFromIndexDTInterval(
+ bucketMicros: Long, k: Long, originMicros: Long, zoneId: ZoneId): Long =
{
+ val bucketOffset = MathUtils.multiplyExact(k, bucketMicros)
+ if (bucketMicros / MICROS_PER_DAY != 0) {
+ timestampAddDayTime(originMicros, bucketOffset, zoneId)
+ } else {
+ // Sub-day bucket stays zone-independent even when k*bucketMicros spans
days.
+ MathUtils.addExact(originMicros, bucketOffset)
+ }
+ }
+
+ /**
+ * The DayTimeInterval bucket containing `tsMicros`, as `(index,
startBoundary)`. The index is
+ * returned so callers walking to the next bucket need not search again.
+ *
+ * `bucketMicros` must be positive; `TimeBucket.checkInputDataTypes`
enforces this at
+ * analysis time.
+ *
+ * @param bucketMicros bucket size in microseconds.
+ * @param tsMicros timestamp to bucket, in microseconds since the epoch
(UTC).
+ * @param originMicros grid alignment anchor, in microseconds since the
epoch (UTC).
+ * @param zoneId zone in which calendar-day arithmetic is performed.
+ */
+ def timeBucketFromTimestampDTInterval(
+ bucketMicros: Long, tsMicros: Long, originMicros: Long, zoneId: ZoneId):
(Long, Long) = {
val diff = MathUtils.subtractExact(tsMicros, originMicros)
- var k = MathUtils.floorDiv(diff, bucketMicros)
+ val k = MathUtils.floorDiv(diff, bucketMicros)
- if (bucketDays == 0) {
- val bucketOffset = MathUtils.multiplyExact(k, bucketMicros)
- MathUtils.addExact(originMicros, bucketOffset)
- } else {
- // bucketMicros >= MICROS_PER_DAY, so DST offset shifts (a few hours at
most) can
- // move candidate(k) within one bucket of `origin + k*bucketMicros` but
no further.
- // One +/-1 step recovers the correct k.
- def candidate(kk: Long): Long =
- timestampAddDayTime(originMicros, MathUtils.multiplyExact(kk,
bucketMicros), zoneId)
-
- var c = candidate(k)
- if (c > tsMicros) {
- k -= 1
- c = candidate(k)
- } else {
- val cNext = candidate(MathUtils.addExact(k, 1L))
- if (cNext <= tsMicros) c = cNext
+ def boundary(kk: Long): Long =
+ timeBucketFromIndexDTInterval(bucketMicros, kk, originMicros, zoneId)
+
+ val bucketStart = boundary(k)
+ if (bucketMicros / MICROS_PER_DAY != 0) {
+ // Multi-day: a DST offset shift moves the boundary by less than one
bucket, so one +/-1 step
+ // recovers the correct index.
+ if (bucketStart > tsMicros) {
+ return (k - 1, boundary(k - 1))
+ }
+ val nextStart = boundary(MathUtils.addExact(k, 1L))
+ if (nextStart <= tsMicros) {
Review Comment:
Good catch, fixed. Replaced the single step with a loop that walks until
boundary(k) <= ts < boundary(k + 1), and added Apia regression tests for epoch
and 1900 origins.
--
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]
