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 234dd7cad0cf [SPARK-58139][SQL] Improve BIN BY test coverage for DST
and bin alignment
234dd7cad0cf is described below
commit 234dd7cad0cfb017a6701fbc9194d65d09c527e0
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Thu Jul 16 21:34:46 2026 +0800
[SPARK-58139][SQL] Improve BIN BY test coverage for DST and bin alignment
### What changes were proposed in this pull request?
This PR adds `BinBySuite` execution tests for day-time interval edge cases,
and reorders the suite to group tests by topic. New tests:
- A DST fall-back bin (the 25-hour day gets a larger ratio share than an
adjacent 24-hour day) and a compound 2-day width across the same transition.
- A sub-day bin in a non-UTC session, exercising the UTC-arithmetic path.
- `ALIGN TO` an origin later than the range, exercising the negative
step-back in the bin-boundary arithmetic.
### Why are the changes needed?
`BinBySuite` covered a DST spring-forward but not the fall-back (a distinct
arithmetic path), a sub-day bin in a non-UTC session, or an origin later than
the range.
### Does this PR introduce _any_ user-facing change?
No. Test-only.
### How was this patch tested?
`BinBySuite` passes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)
Closes #57278 from vranes/bin-by-dst-fallback-tests.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../scala/org/apache/spark/sql/BinBySuite.scala | 221 +++++++++++++++------
1 file changed, 155 insertions(+), 66 deletions(-)
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 3e00a0645e0d..db01411c75f7 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
@@ -34,20 +34,25 @@ class BinBySuite extends QueryTest with SharedSparkSession {
private def ratio(overlapMicros: Long, totalMicros: Long): Double =
overlapMicros.toDouble / totalMicros.toDouble
- private def createMetricsView(): Unit = {
- spark.sql(
- """SELECT TIMESTAMP '2024-01-01 00:00:00' AS ts_start,
- | TIMESTAMP '2024-01-01 01:00:00' AS ts_end,
- | CAST(1 AS DOUBLE) AS
value""".stripMargin).createOrReplaceTempView("metrics")
+ test("BIN BY is rejected when the operator is disabled") {
+ withSQLConf(SQLConf.BIN_BY_ENABLED.key -> "false") {
+ // Disabled, BIN BY is rejected at analysis with
UNSUPPORTED_FEATURE.BIN_BY. spark.sql eagerly
+ // analyzes, so the query itself is inside the intercept.
+ checkError(
+ exception = intercept[SparkThrowable] {
+ spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01
00:10:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | DISTRIBUTE UNIFORM (value))""".stripMargin)
+ },
+ condition = "UNSUPPORTED_FEATURE.BIN_BY",
+ parameters = Map.empty[String, String])
+ }
}
- private val binByQuery =
- """SELECT * FROM metrics BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (value)
- |)""".stripMargin
-
test("BIN BY splits a range into proportional sub-rows") {
withSQLConf(
SQLConf.BIN_BY_ENABLED.key -> "true",
@@ -104,64 +109,73 @@ class BinBySuite extends QueryTest with
SharedSparkSession {
}
}
- test("BIN BY emits a NULL-range row with all computed columns NULL") {
+ test("BIN BY aligns to an origin later than the range") {
withSQLConf(
SQLConf.BIN_BY_ENABLED.key -> "true",
SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ // ALIGN TO after the range exercises a negative bucket index (origin
later than the range).
+ // The grid still lands on :00/:05/:10, so [00:02, 00:12) splits into 3,
5, 2 minutes.
val df = spark.sql(
- """SELECT id, bin_start, bin_end, bin_distribute_ratio, value
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
|FROM VALUES
- | (1, CAST(NULL AS TIMESTAMP), TIMESTAMP '2024-01-01 00:10:00',
100.0D),
- | (2, TIMESTAMP '2024-01-01 00:00:00', CAST(NULL AS TIMESTAMP),
200.0D)
- | AS metrics(id, ts_start, ts_end, value)
+ | (TIMESTAMP '2024-01-01 00:02:00', TIMESTAMP '2024-01-01
00:12:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
|BIN BY (
| RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
- | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))
- |ORDER BY id""".stripMargin)
- // A NULL range nulls every computed column; only the `id` passthrough
survives.
+ | ALIGN TO TIMESTAMP '2024-01-01 00:20:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start""".stripMargin)
+ val m = 60L * 1000000L // one minute in micros
+ val total = 10 * m
checkAnswer(df, Seq(
- Row(1, null, null, null, null),
- Row(2, null, null, null, null)))
+ Row(tsAt("2024-01-01 00:00:00"), tsAt("2024-01-01 00:05:00"),
+ ratio(3 * m, total), 100.0 * ratio(3 * m, total)),
+ Row(tsAt("2024-01-01 00:05:00"), tsAt("2024-01-01 00:10:00"),
+ ratio(5 * m, total), 100.0 * ratio(5 * m, total)),
+ Row(tsAt("2024-01-01 00:10:00"), tsAt("2024-01-01 00:15:00"),
+ ratio(2 * m, total), 100.0 * ratio(2 * m, total))))
}
}
- test("BIN BY raises BIN_BY_INVALID_RANGE for an inverted range") {
+ test("BIN BY replicates a nested struct passthrough column across a
multi-bin split") {
withSQLConf(
SQLConf.BIN_BY_ENABLED.key -> "true",
SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ // A nested struct passthrough must appear identically on every split
sub-row.
val df = spark.sql(
- """SELECT * FROM VALUES
- | (TIMESTAMP '2024-01-01 00:10:00', TIMESTAMP '2024-01-01
00:00:00', 100.0D)
- | AS metrics(ts_start, ts_end, value)
+ """SELECT s, bin_start, value
+ |FROM VALUES
+ | (named_struct('a', 1, 'b', 'x'),
+ | TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01
00:10:00', 100.0D)
+ | AS metrics(s, ts_start, ts_end, value)
|BIN BY (
| RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
- | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))""".stripMargin)
- checkError(
- exception = intercept[SparkThrowable] {
- df.collect()
- },
- condition = "BIN_BY_INVALID_RANGE",
- parameters = Map(
- "rangeStart" -> "2024-01-01 00:10:00",
- "rangeEnd" -> "2024-01-01 00:00:00"))
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(Row(1, "x"), tsAt("2024-01-01 00:00:00"), 50.0),
+ Row(Row(1, "x"), tsAt("2024-01-01 00:05:00"), 50.0)))
}
}
- test("BIN BY emits a single ratio-1.0 row for a zero-length range") {
+ test("BIN BY uses UTC arithmetic for a sub-day bin in a non-UTC session") {
+ val la = ZoneId.of("America/Los_Angeles")
withSQLConf(
SQLConf.BIN_BY_ENABLED.key -> "true",
- SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+ // Sub-day widths use UTC microsecond arithmetic, not the civil-time
path the multi-day tests
+ // exercise; the 5-minute grid lands on clean boundaries even in a
non-UTC session.
val df = spark.sql(
"""SELECT bin_start, bin_end, bin_distribute_ratio, value
|FROM VALUES
- | (TIMESTAMP '2024-01-01 00:02:00', TIMESTAMP '2024-01-01
00:02:00', 100.0D)
+ | (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01
00:10:00', 100.0D)
| AS metrics(ts_start, ts_end, value)
|BIN BY (
| RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
- | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))""".stripMargin)
- // rangeStart == rangeEnd: one row, ratio 1.0, value kept.
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start""".stripMargin)
checkAnswer(df, Seq(
- Row(tsAt("2024-01-01 00:00:00"), tsAt("2024-01-01 00:05:00"), 1.0,
100.0)))
+ Row(tsAt("2024-01-01 00:00:00", la), tsAt("2024-01-01 00:05:00", la),
0.5, 50.0),
+ Row(tsAt("2024-01-01 00:05:00", la), tsAt("2024-01-01 00:10:00", la),
0.5, 50.0)))
}
}
@@ -193,46 +207,60 @@ class BinBySuite extends QueryTest with
SharedSparkSession {
}
}
- test("BIN BY replicates a nested struct passthrough column across a
multi-bin split") {
+ test("BIN BY uses civil-time bin boundaries across a DST fall-back") {
+ val la = ZoneId.of("America/Los_Angeles")
withSQLConf(
SQLConf.BIN_BY_ENABLED.key -> "true",
- SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
- // A nested struct passthrough must appear identically on every split
sub-row.
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+ // A 1-DAY bin spanning the 2024-11-03 fall-back: the 2024-11-03 bin is
25h wide (01:00 PDT
+ // repeats as 01:00 PST) while 2024-11-04 is 24h, so it gets a larger
share of the range.
val df = spark.sql(
- """SELECT s, bin_start, value
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
|FROM VALUES
- | (named_struct('a', 1, 'b', 'x'),
- | TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01
00:10:00', 100.0D)
- | AS metrics(s, ts_start, ts_end, value)
+ | (TIMESTAMP '2024-11-03 00:00:00', TIMESTAMP '2024-11-05
00:00:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
|BIN BY (
- | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
- | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '1' DAY
+ | ALIGN TO TIMESTAMP '2024-11-03 00:00:00' DISTRIBUTE UNIFORM
(value))
|ORDER BY bin_start""".stripMargin)
+ val h = 3600L * 1000000L // one hour in micros
+ val total = 49 * h // 25h (DST day) + 24h
checkAnswer(df, Seq(
- Row(Row(1, "x"), tsAt("2024-01-01 00:00:00"), 50.0),
- Row(Row(1, "x"), tsAt("2024-01-01 00:05:00"), 50.0)))
+ Row(tsAt("2024-11-03 00:00:00", la), tsAt("2024-11-04 00:00:00", la),
+ ratio(25 * h, total), 100.0 * ratio(25 * h, total)),
+ Row(tsAt("2024-11-04 00:00:00", la), tsAt("2024-11-05 00:00:00", la),
+ ratio(24 * h, total), 100.0 * ratio(24 * h, total))))
}
}
- test("BIN BY is rejected when the operator is disabled") {
- withSQLConf(SQLConf.BIN_BY_ENABLED.key -> "false") {
- withTempView("metrics") {
- createMetricsView()
- // Disabled, the operator is rejected at analysis with the same
UNSUPPORTED_FEATURE.BIN_BY
- // condition the execution stub raises when enabled.
- checkError(
- exception = intercept[SparkThrowable] {
- spark.sql(binByQuery).queryExecution.assertAnalyzed()
- },
- condition = "UNSUPPORTED_FEATURE.BIN_BY",
- parameters = Map.empty[String, String])
- }
+ test("BIN BY uses civil-time boundaries for a compound multi-day width
across DST") {
+ val la = ZoneId.of("America/Los_Angeles")
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+ // A 2-DAY bin across the 2024-11-03 fall-back: boundaries land two
civil days apart, so the
+ // first bin (Nov 2 + Nov 3) is 24h + 25h = 49h and the second (Nov 4 +
Nov 5) is 48h.
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP '2024-11-02 00:00:00', TIMESTAMP '2024-11-06
00:00:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '2' DAY
+ | ALIGN TO TIMESTAMP '2024-11-02 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start""".stripMargin)
+ val h = 3600L * 1000000L // one hour in micros
+ val total = 97 * h // 49h (Nov 2 + fall-back Nov 3) + 48h (Nov 4 +
Nov 5)
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-11-02 00:00:00", la), tsAt("2024-11-04 00:00:00", la),
+ ratio(49 * h, total), 100.0 * ratio(49 * h, total)),
+ Row(tsAt("2024-11-04 00:00:00", la), tsAt("2024-11-06 00:00:00", la),
+ ratio(48 * h, total), 100.0 * ratio(48 * h, total))))
}
}
test("BIN BY executes on NTZ inputs with the epoch default origin") {
withSQLConf(SQLConf.BIN_BY_ENABLED.key -> "true") {
- // NTZ inputs default the origin to the wall-clock epoch.
val df = spark.sql(
"""SELECT bin_start, bin_end, bin_distribute_ratio, value
|FROM VALUES
@@ -266,4 +294,65 @@ class BinBySuite extends QueryTest with SharedSparkSession
{
Row(ntz("2024-03-11 00:00:00"), ntz("2024-03-12 00:00:00"), 0.5,
50.0)))
}
}
+
+ test("BIN BY emits a single ratio-1.0 row for a zero-length range") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:02:00', TIMESTAMP '2024-01-01
00:02:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))""".stripMargin)
+ // rangeStart == rangeEnd: one row, ratio 1.0, value kept.
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-01-01 00:00:00"), tsAt("2024-01-01 00:05:00"), 1.0,
100.0)))
+ }
+ }
+
+ test("BIN BY raises BIN_BY_INVALID_RANGE for an inverted range") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val df = spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:10:00', TIMESTAMP '2024-01-01
00:00:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))""".stripMargin)
+ checkError(
+ exception = intercept[SparkThrowable] {
+ df.collect()
+ },
+ condition = "BIN_BY_INVALID_RANGE",
+ parameters = Map(
+ "rangeStart" -> "2024-01-01 00:10:00",
+ "rangeEnd" -> "2024-01-01 00:00:00"))
+ }
+ }
+
+ test("BIN BY emits a NULL-range row with all computed columns NULL") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val df = spark.sql(
+ """SELECT id, bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (1, CAST(NULL AS TIMESTAMP), TIMESTAMP '2024-01-01 00:10:00',
100.0D),
+ | (2, TIMESTAMP '2024-01-01 00:00:00', CAST(NULL AS TIMESTAMP),
200.0D)
+ | AS metrics(id, ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY id""".stripMargin)
+ // A NULL range nulls every computed column; only the `id` passthrough
survives.
+ checkAnswer(df, Seq(
+ Row(1, null, null, null, null),
+ Row(2, null, null, null, null)))
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
