cloud-fan commented on code in PR #57401:
URL: https://github.com/apache/spark/pull/57401#discussion_r3630614608
##########
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:
The containing-bucket helper must establish `boundary(k) <= ts < boundary(k
+ 1)`, but a single correction step does not always do that. For `Pacific/Apia`
with a one-day bucket, origin `1900-01-01 00:00`, and timestamp `2012-01-02
00:00`, the raw index is 40906 and both boundaries 40907 and 40908 are at or
before the timestamp, so this returns the previous civil day. Please adjust
until the full containment invariant holds and add this distant-origin case as
a regression test.
--
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]
