This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 07144c0bfcb2 [SPARK-58237][SQL] Fix BIN BY multi-day civil-time bin
boundaries to use the absolute grid
07144c0bfcb2 is described below
commit 07144c0bfcb2a80b77a93631121ab31a743675de
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Thu Jul 23 21:54:44 2026 +0800
[SPARK-58237][SQL] Fix BIN BY multi-day civil-time bin boundaries to use
the absolute grid
### What changes were proposed in this pull request?
`BinByExec` computed multi-day civil-time bin boundaries by forward-walking
(each bin's end reused as the next bin's start) instead of placing every
boundary on the absolute grid `bin_start(k) = timestampAddDayTime(origin, k *
width)`, the grid `time_bucket` uses. This PR switches it to the absolute grid:
- `DateTimeUtils.timeBucketDTInterval` is factored into
`timeBucketFromIndexDTInterval` (the grid boundary of a bucket index) and
`timeBucketFromTimestampDTInterval` (the `(index, startBoundary)` of the bucket
containing a timestamp), with `timeBucketDTInterval` now delegating to the
latter. Behavior-preserving for `time_bucket`.
- `BinByExec` advances the bucket index and reads each boundary off the
grid via `timeBucketFromIndexDTInterval`, instead of walking with
`timestampAddDayTime`.
### Why are the changes needed?
`timestampAddDayTime` adds the width's day part as calendar days
(DST-absorbing: a day is 23h or 25h across a transition) and the sub-day
remainder as real microseconds, so it is not associative across a DST/offset
transition:
```
addDayTime(addDayTime(origin, k*W), W) != addDayTime(origin, (k+1)*W)
```
So for a width with a sub-day part (e.g. `INTERVAL '36' HOUR`) spanning a
DST transition, the walk drifts off the grid by the transition hour from the
second bin onward. The same calendar bin then gets a different
`bin_start`/`bin_end` depending on where a row's range began, fragmenting
`GROUP BY bin_start`. Whole-day widths, sub-day widths, and `TIMESTAMP_NTZ` are
unaffected.
### Does this PR introduce _any_ user-facing change?
No, this is a bug fix for the BIN BY operator which is gated off by
default. Second-and-later bin boundaries of a range spanning a DST transition,
under a non-whole-day width in a non-UTC session, now land on the absolute grid
instead of drifting by the transition hour.
### How was this patch tested?
- `DateTimeUtilsSuite`: existing `timeBucketDTInterval` cases unchanged
(regression guard for the factoring), plus a round-trip test that the
containing bucket agrees with `timeBucketDTInterval` and encloses the timestamp.
- `BinBySuite`: new tests for multi-bin 36h grid boundaries across a
spring-forward, boundary agreement between rows reaching a bin from different
starts, a `Pacific/Apia` whole-day skip, and a zero-length range. Each fails on
the old walk and passes after the fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)
Closes #57401 from vranes/bin-by-ltz-absolute-grid.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/sql/catalyst/util/DateTimeUtils.scala | 93 ++++++++++++------
.../sql/catalyst/util/DateTimeUtilsSuite.scala | 48 ++++++++++
.../org/apache/spark/sql/execution/BinByExec.scala | 21 +++--
.../scala/org/apache/spark/sql/BinBySuite.scala | 105 +++++++++++++++++++++
4 files changed, 230 insertions(+), 37 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
index 2fe2e5f4c988..24aabdc988fb 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
@@ -1266,47 +1266,82 @@ 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
-
- val diff = MathUtils.subtractExact(tsMicros, originMicros)
- var k = MathUtils.floorDiv(diff, bucketMicros)
+ def timeBucketFromIndexDTInterval(
+ bucketMicros: Long, k: Long, originMicros: Long, zoneId: ZoneId): Long =
{
+ val bucketOffset = MathUtils.multiplyExact(k, bucketMicros)
- if (bucketDays == 0) {
- val bucketOffset = MathUtils.multiplyExact(k, bucketMicros)
- MathUtils.addExact(originMicros, bucketOffset)
+ if (bucketMicros / MICROS_PER_DAY != 0) {
+ timestampAddDayTime(originMicros, bucketOffset, zoneId)
} 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
+ // 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)
+ val k = MathUtils.floorDiv(diff, bucketMicros)
+
+ def boundary(kk: Long): Long =
+ timeBucketFromIndexDTInterval(bucketMicros, kk, originMicros, zoneId)
+
+ if (bucketMicros / MICROS_PER_DAY != 0) {
+ // Multi-day: calendar-day arithmetic makes boundaries drift off the
linear grid as offset
+ // shifts accumulate between origin and ts. A DST shift (e.g.
Pacific/Los_Angeles) is 1h, so
+ // the estimate is off by at most one; a date-line shift (e.g.
Pacific/Apia 2011) can reach a
+ // full bucket, so it can be off by more than one. Step the index until
the bucket contains
+ // ts: the first loop walks down while the start is past ts, the second
walks up while the
+ // next start is not, landing on boundary(k) <= ts < boundary(k + 1).
+ var adjusted = k
+ while (boundary(adjusted) > tsMicros) {
+ adjusted = MathUtils.subtractExact(adjusted, 1L)
}
- c
+ while (boundary(MathUtils.addExact(adjusted, 1L)) <= tsMicros) {
+ adjusted = MathUtils.addExact(adjusted, 1L)
+ }
+ return (adjusted, boundary(adjusted))
}
+ // Sub-day: pure UTC arithmetic, the linear estimate is exact.
+ (k, boundary(k))
+ }
+
+ /**
+ * DayTimeInterval bucketing: the start boundary of the bucket containing
`tsMicros`.
+ *
+ * @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 timeBucketDTInterval(
+ bucketMicros: Long, tsMicros: Long, originMicros: Long, zoneId: ZoneId):
Long = {
+ val (_, start) = timeBucketFromTimestampDTInterval(bucketMicros, tsMicros,
originMicros, zoneId)
+ start
}
/**
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala
index d07265c26e97..196d32e1a4f9 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala
@@ -2051,6 +2051,54 @@ class DateTimeUtilsSuite extends SparkFunSuite with
Matchers with SQLHelper {
}
}
+ test("timeBucketFromIndexDTInterval / timeBucketFromTimestampDTInterval") {
+ val utc = ZoneOffset.UTC
+ val la = DateTimeUtils.getZoneId("America/Los_Angeles")
+
+ // The containing bucket agrees with timeBucketDTInterval and encloses ts:
+ // boundary(k) == start <= ts < boundary(k + 1).
+ def checkContaining(bucket: Long, ts: Long, origin: Long, zone: ZoneId):
Unit = {
+ val (k, start) = timeBucketFromTimestampDTInterval(bucket, ts, origin,
zone)
+ assert(start === timeBucketFromIndexDTInterval(bucket, k, origin, zone))
+ assert(start === timeBucketDTInterval(bucket, ts, origin, zone))
+ assert(start <= ts)
+ assert(ts < timeBucketFromIndexDTInterval(bucket, k + 1, origin, zone))
+ }
+ // Sub-day, sub-day with custom origin, and multi-day (36h) across both LA
DST transitions.
+ checkContaining(15 * MICROS_PER_MINUTE, date(2024, 1, 1, 11, 27, 0), 0L,
la)
+ checkContaining(MICROS_PER_HOUR, date(2024, 1, 1, 11, 27, 0), date(1970,
1, 1, 0, 5, 0), utc)
+ val laOrigin = date(2024, 3, 9, 8, 0, 0) // 08:00 UTC = 2024-03-09 00:00
PST
+ checkContaining(36 * MICROS_PER_HOUR, date(2024, 3, 12, 9, 0, 0),
laOrigin, la)
+ checkContaining(MICROS_PER_DAY, date(2024, 11, 4, 2, 0, 0), date(2024, 11,
1, 7, 0, 0), la)
+
+ // Pacific/Apia's 2011 date-line shift throws the 1-day estimate off by
one bucket (epoch
+ // origin) or two (1900 origin), so the search must step more than once.
+ val apiaZone = DateTimeUtils.getZoneId("Pacific/Apia")
+ checkContaining(MICROS_PER_DAY, date(2012, 1, 2, 0, 0, 0, 0, apiaZone),
+ date(1970, 1, 1), apiaZone)
+ checkContaining(MICROS_PER_DAY, date(2012, 1, 2, 0, 0, 0, 0, apiaZone),
+ date(1900, 1, 1, 0, 0, 0, 0, apiaZone), apiaZone)
+
+ // Boundaries land on the grid across the spring-forward, not 1h off as a
walk would drift.
+ val (idx, _) = timeBucketFromTimestampDTInterval(36 * MICROS_PER_HOUR,
+ date(2024, 3, 12, 9, 0, 0), laOrigin, la)
+ assert(timeBucketFromIndexDTInterval(36 * MICROS_PER_HOUR, idx, laOrigin,
la)
+ === date(2024, 3, 12, 7, 0, 0)) // 2024-03-12 00:00 PDT, the grid
(not walked) value
+ assert(timeBucketFromIndexDTInterval(36 * MICROS_PER_HOUR, idx + 1,
laOrigin, la)
+ === date(2024, 3, 13, 19, 0, 0)) // 2024-03-13 12:00 PDT
+
+ // Whole-day zone skip: Apia skipped 2011-12-30, so two consecutive 1-day
grid boundaries
+ // collapse to the same instant (civil "2011-12-30 00:00" and "2011-12-31
00:00" are identical).
+ val apia = DateTimeUtils.getZoneId("Pacific/Apia")
+ val apiaOrigin = date(2011, 12, 25, 10, 0, 0) // 2011-12-25 00:00 Apia
(UTC-11 -> 10:00 UTC)
+ // This ts lands in the bucket whose start collapsed onto the previous
one's, so
+ // boundary(k - 1) == boundary(k).
+ val (collapsed, _) = timeBucketFromTimestampDTInterval(MICROS_PER_DAY,
+ date(2011, 12, 30, 12, 0, 0), apiaOrigin, apia)
+ assert(timeBucketFromIndexDTInterval(MICROS_PER_DAY, collapsed - 1,
apiaOrigin, apia)
+ === timeBucketFromIndexDTInterval(MICROS_PER_DAY, collapsed, apiaOrigin,
apia))
+ }
+
test("timeBucketYMInterval") {
val utc = ZoneOffset.UTC
// 1-month bucket default origin
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/BinByExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/BinByExec.scala
index 22898f6dfdbd..d44092ac2c9e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/BinByExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/BinByExec.scala
@@ -33,9 +33,9 @@ import org.apache.spark.sql.types.DoubleType
* bin overlapping `[rangeStart, rangeEnd)`, scaling the DISTRIBUTE UNIFORM
columns by the overlap
* fraction and appending `bin_start`, `bin_end`, `bin_distribute_ratio`.
*
- * Bin boundaries reuse [[DateTimeUtils.timeBucketDTInterval]] /
- * [[DateTimeUtils.timestampAddDayTime]], matching `time_bucket`: sub-day
widths use UTC microsecond
- * arithmetic, multi-day widths use civil-time arithmetic in the session zone
(UTC for
+ * Bin boundaries lie on the grid `bin_start(k) =
+ * [[DateTimeUtils.timeBucketFromIndexDTInterval]](width, k, origin, zone)`.
Sub-day widths use UTC
+ * microsecond arithmetic, multi-day widths use civil-time arithmetic in the
session zone (UTC for
* TIMESTAMP_NTZ). DISTRIBUTE UNIFORM columns are FLOAT or DOUBLE only and
scale by multiplication.
*/
case class BinByExec(
@@ -107,8 +107,9 @@ case class BinByExec(
if (rs > re) {
throw
QueryExecutionErrors.binByInvalidRangeError(fmt.format(rs), fmt.format(re))
} else if (rs == re) {
- val binStart = DateTimeUtils.timeBucketDTInterval(width, rs,
origin, zone)
- val binEnd = DateTimeUtils.timestampAddDayTime(binStart, width,
zone)
+ val (k, binStart) =
DateTimeUtils.timeBucketFromTimestampDTInterval(width, rs, origin,
+ zone)
+ val binEnd = DateTimeUtils.timeBucketFromIndexDTInterval(width,
k + 1, origin, zone)
appended.update(0, binStart)
appended.update(1, binEnd)
appended.update(2, 1.0d)
@@ -116,17 +117,21 @@ case class BinByExec(
} else {
val total = Math.subtractExact(re, rs)
new Iterator[InternalRow] {
- private var curStart =
- DateTimeUtils.timeBucketDTInterval(width, rs, origin, zone)
+ // Advance the bucket index and read each boundary off the
grid, rather than
+ // walking one bin end to the next start.
+ private var (k, curStart) =
+ DateTimeUtils.timeBucketFromTimestampDTInterval(width, rs,
origin, zone)
override def hasNext: Boolean = curStart < re
override def next(): InternalRow = {
- val curEnd = DateTimeUtils.timestampAddDayTime(curStart,
width, zone)
+ val curEnd =
+ DateTimeUtils.timeBucketFromIndexDTInterval(width, k + 1,
origin, zone)
val overlap = math.min(re, curEnd) - math.max(rs, curStart)
appended.update(0, curStart)
appended.update(1, curEnd)
appended.update(2, overlap.toDouble / total.toDouble)
+ k += 1
curStart = curEnd
proj(joined.withRight(appended))
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
index db01411c75f7..7ec687e3d953 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
@@ -259,6 +259,111 @@ class BinBySuite extends QueryTest with
SharedSparkSession {
}
}
+ test("BIN BY places multi-bin civil-time boundaries on the absolute grid
across DST") {
+ val la = ZoneId.of("America/Los_Angeles")
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+ // Boundaries of a 36h width must lie on the grid bin_start(k) =
ALIGN_TO + k * 36h. Across
+ // the 2024-03-10 spring-forward, a forward walk would drift 1h off the
grid from bin 1 on.
+ // This range spans four bins.
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP '2024-03-09 06:00:00', TIMESTAMP '2024-03-13
18:00:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '36' HOUR
+ | ALIGN TO TIMESTAMP '2024-03-09 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start""".stripMargin)
+ val h = 3600L * 1000000L
+ val total = 107 * h // 30h + 35h + 36h + 6h
+ // Bin 2 starts at 2024-03-12 00:00 LA (07:00 UTC); a forward walk would
report 08:00 UTC.
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-03-09 00:00:00", la), tsAt("2024-03-10 13:00:00", la),
+ ratio(30 * h, total), 100.0 * ratio(30 * h, total)),
+ Row(tsAt("2024-03-10 13:00:00", la), tsAt("2024-03-12 00:00:00", la),
+ ratio(35 * h, total), 100.0 * ratio(35 * h, total)),
+ Row(tsAt("2024-03-12 00:00:00", la), tsAt("2024-03-13 12:00:00", la),
+ ratio(36 * h, total), 100.0 * ratio(36 * h, total)),
+ Row(tsAt("2024-03-13 12:00:00", la), tsAt("2024-03-15 00:00:00", la),
+ ratio(6 * h, total), 100.0 * ratio(6 * h, total))))
+ }
+ }
+
+ test("BIN BY reports the same civil-time bin boundary regardless of where a
row starts") {
+ val la = ZoneId.of("America/Los_Angeles")
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+ // Both rows span the bin starting 2024-03-12 00:00 LA: one walks into
it from bin 0, the
+ // other starts inside it. Both must report the same bin_start; a
forward walk would report it
+ // an hour apart, splitting GROUP BY bin_start.
+ val df = spark.sql(
+ """SELECT ts_start, bin_start, bin_end
+ |FROM VALUES
+ | (TIMESTAMP '2024-03-09 06:00:00', TIMESTAMP '2024-03-13
06:00:00', 1.0D),
+ | (TIMESTAMP '2024-03-12 03:00:00', TIMESTAMP '2024-03-13
06:00:00', 1.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '36' HOUR
+ | ALIGN TO TIMESTAMP '2024-03-09 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |WHERE bin_start = TIMESTAMP '2024-03-12 00:00:00'
+ |ORDER BY ts_start""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-03-09 06:00:00", la), tsAt("2024-03-12 00:00:00", la),
+ tsAt("2024-03-13 12:00:00", la)),
+ Row(tsAt("2024-03-12 03:00:00", la), tsAt("2024-03-12 00:00:00", la),
+ tsAt("2024-03-13 12:00:00", la))))
+ }
+ }
+
+ test("BIN BY emits a zero-width bin at a whole-day zone skip") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "Pacific/Apia") {
+ // Apia skipped all of 2011-12-30 (UTC-11 -> UTC+13), so two grid
boundaries land on the same
+ // instant, emitting a zero-width ratio-0 bin (bin_start == bin_end).
The real bins still tile
+ // the range, so ratios sum to 1.0. Compared as UTC instants: a civil
label on the skipped day
+ // is ambiguous.
+ val utc = ZoneOffset.UTC
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP '2011-12-29 12:00:00', TIMESTAMP '2011-12-31
12:00:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '1' DAY
+ | ALIGN TO TIMESTAMP '2011-12-25 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start, bin_end""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(tsAt("2011-12-29 10:00:00", utc), tsAt("2011-12-30 10:00:00",
utc), 0.5, 50.0),
+ Row(tsAt("2011-12-30 10:00:00", utc), tsAt("2011-12-30 10:00:00",
utc), 0.0, 0.0),
+ Row(tsAt("2011-12-30 10:00:00", utc), tsAt("2011-12-31 10:00:00",
utc), 0.5, 50.0)))
+ }
+ }
+
+ test("BIN BY places the zero-length range bin on the grid for a
non-whole-day width") {
+ val la = ZoneId.of("America/Los_Angeles")
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+ // A zero-length range emits one ratio-1.0 row for its bin. Its bin's
36h step crosses the
+ // spring-forward, so bin_end must be the grid boundary (2024-03-12
00:00), not the walked
+ // bin_start + 36h (2024-03-12 01:00).
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP '2024-03-11 12:00:00', TIMESTAMP '2024-03-11
12:00:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '36' HOUR
+ | ALIGN TO TIMESTAMP '2024-03-09 00:00:00' DISTRIBUTE UNIFORM
(value))""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-03-10 13:00:00", la), tsAt("2024-03-12 00:00:00", la),
1.0, 100.0)))
+ }
+ }
+
test("BIN BY executes on NTZ inputs with the epoch default origin") {
withSQLConf(SQLConf.BIN_BY_ENABLED.key -> "true") {
val df = spark.sql(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
