MaxGekk commented on code in PR #53237:
URL: https://github.com/apache/spark/pull/53237#discussion_r3482684252
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/timeExpressions.scala:
##########
@@ -752,6 +752,111 @@ case class TimeTrunc(unit: Expression, time: Expression)
}
}
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = """
+ _FUNC_(bucket_width, time) - Returns the start of the time bucket
containing `time`,
+ where buckets are aligned to midnight (00:00:00).
+ """,
+ arguments = """
+ Arguments:
+ * bucket_width - A day-time interval specifying the width of each bucket
(e.g., INTERVAL '15' MINUTE).
+ * time - The time value to bucket. Must be of TIME type.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_(INTERVAL '15' MINUTE, TIME'09:37:22');
+ 09:30:00
+ > SELECT _FUNC_(INTERVAL '30' MINUTE, TIME'14:47:00');
+ 14:30:00
+ > SELECT _FUNC_(INTERVAL '1' HOUR, TIME'16:35:00');
+ 16:00:00
+ > SELECT _FUNC_(INTERVAL '2' HOUR, TIME'15:20:00');
+ 14:00:00
+ """,
+ note = """
+ Notes:
+ * Buckets are aligned to midnight (00:00:00) and cannot span across
midnight.
+ * The bucket width must be a positive day-time interval.
+ * The function returns the start time of the bucket (inclusive lower
bound).
+ """,
+ group = "datetime_funcs",
+ since = "4.2.0")
+// scalastyle:on line.size.limit
+case class TimeBucket(
+ bucketWidth: Expression,
+ time: Expression)
+ extends BinaryExpression with ImplicitCastInputTypes {
+
+ override def left: Expression = bucketWidth
+ override def right: Expression = time
+
+ override def inputTypes: Seq[AbstractDataType] =
+ Seq(DayTimeIntervalType, AnyTimeType)
+
+ override def dataType: DataType = time.dataType
+
+ override def nullable: Boolean = time.nullable || bucketWidth.nullable
+
+ override def nullIntolerant: Boolean = true
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ super.checkInputDataTypes() match {
+ case TypeCheckSuccess =>
+ if (!bucketWidth.foldable) {
+ DataTypeMismatch(
+ errorSubClass = "NON_FOLDABLE_INPUT",
+ messageParameters = Map(
+ "inputName" -> toSQLId("bucket_width"),
+ "inputType" -> toSQLType(bucketWidth.dataType),
+ "inputExpr" -> toSQLExpr(bucketWidth)
+ )
+ )
+ } else {
+ val widthValue = bucketWidth.eval()
+ if (widthValue != null) {
+ val widthMicros = widthValue.asInstanceOf[Long]
+ if (widthMicros <= 0) {
+ DataTypeMismatch(
+ errorSubClass = "VALUE_OUT_OF_RANGE",
+ messageParameters = Map(
+ "exprName" -> "bucket_width",
+ "valueRange" -> s"(0, ${Long.MaxValue}]",
+ "currentValue" -> toSQLValue(widthMicros, LongType)
+ )
+ )
+ } else {
+ TypeCheckSuccess
+ }
+ } else {
+ TypeCheckSuccess
+ }
+ }
+ case failure => failure
+ }
+ }
+
+ override def nullSafeEval(bucketWidthValue: Any, timeValue: Any): Any = {
+ val bucketMicros = bucketWidthValue.asInstanceOf[Long]
+ val timeNanos = timeValue.asInstanceOf[Long]
+ val bucketNanos = bucketMicros * NANOS_PER_MICROS
Review Comment:
Two boundary issues here. (1) `checkInputDataTypes` only enforces `width >
0`, but the `@note` says buckets "cannot span across midnight". A width ≥ 24h
is accepted and degenerate: `bucketNanos ≥ max(timeNanos)`, so every TIME
divides to bucket 0 and returns `00:00:00` (e.g. `INTERVAL '25' HOUR`). (2)
`bucketMicros * NANOS_PER_MICROS` (and `$bucket * 1000L` in `doGenCode`) is a
raw multiply, unlike `SparkDateTimeUtils.microsToNanos`, which uses
`Math.multiplyExact`; a very large interval silently overflows `Long` → a
garbage/negative `bucketNanos` and a wrong result with no error. Suggest
validating `bucket_width ∈ (0, 24h]` in `checkInputDataTypes` (matching the
documented constraint) and scaling via `microsToNanos` / `Math.multiplyExact`.
##########
sql/api/src/main/scala/org/apache/spark/sql/functions.scala:
##########
@@ -6910,6 +6910,36 @@ object functions {
Column.fn("time_trunc", unit, time)
}
+ /**
+ * Returns the start of the time bucket containing the input time value.
Buckets are aligned to
+ * midnight (00:00:00).
+ *
+ * @param bucketWidth
+ * A Column representing a day-time interval (e.g., INTERVAL '15' MINUTE)
+ * @param time
+ * The time column to bucket
+ *
+ * @group datetime_funcs
+ * @since 4.2.0
+ */
+ def time_bucket(bucketWidth: Column, time: Column): Column =
+ Column.fn("time_bucket", bucketWidth, time)
+
+ /**
+ * Returns the start of the time bucket containing the input time value.
Buckets are aligned to
+ * midnight (00:00:00).
+ *
+ * @param bucketWidth
+ * A string representing a day-time interval (e.g., "15 minutes")
+ * @param time
+ * The time column to bucket
+ *
+ * @group datetime_funcs
+ * @since 4.2.0
+ */
+ def time_bucket(bucketWidth: String, time: Column): Column =
+ time_bucket(expr(s"INTERVAL '$bucketWidth'"), time)
Review Comment:
This builds the interval by interpolating the caller's string straight into
a parsed SQL fragment — the only `expr(s"INTERVAL '…'")` in `functions.scala`.
A string with a quote or an expression breaks parsing or yields an unintended
interval. The `window(_, windowDuration: String)` overload instead passes the
duration as a literal parsed safely inside `TimeWindow`; mirroring that here
would be more robust than string-building SQL. (This overload is also currently
untested.)
##########
sql/core/src/test/scala/org/apache/spark/sql/TimeFunctionsSuiteBase.scala:
##########
@@ -478,6 +478,159 @@ abstract class TimeFunctionsSuiteBase extends QueryTest
with SharedSparkSession
)
}
+ test("time_bucket with various intervals") {
+ // 15-minute buckets
+ checkAnswer(
+ sql("SELECT time_bucket(INTERVAL '15' MINUTE, TIME'09:37:22')"),
+ Row(LocalTime.of(9, 30, 0)) :: Nil)
+
+ // 30-minute buckets
+ checkAnswer(
+ sql("SELECT time_bucket(INTERVAL '30' MINUTE, TIME'14:47:00')"),
+ Row(LocalTime.of(14, 30, 0)) :: Nil)
+
+ // 1-hour buckets
+ checkAnswer(
+ sql("SELECT time_bucket(INTERVAL '1' HOUR, TIME'16:35:00')"),
+ Row(LocalTime.of(16, 0, 0)) :: Nil)
+
+ // 2-hour buckets
+ checkAnswer(
+ sql("SELECT time_bucket(INTERVAL '2' HOUR, TIME'15:20:00')"),
+ Row(LocalTime.of(14, 0, 0)) :: Nil)
+
+ // Complex interval (90 minutes = 1.5 hours)
+ checkAnswer(
+ sql("SELECT time_bucket(INTERVAL '90' MINUTE, TIME'10:45:00')"),
+ Row(LocalTime.of(10, 30, 0)) :: Nil)
+ }
+
+ test("time_bucket edge cases") {
+ // Midnight (start of day)
+ checkAnswer(
+ sql("SELECT time_bucket(INTERVAL '1' HOUR, TIME'00:00:00')"),
+ Row(LocalTime.of(0, 0, 0)) :: Nil)
+
+ // End of day
+ checkAnswer(
+ sql("SELECT time_bucket(INTERVAL '30' MINUTE, TIME'23:59:59.999999')"),
+ Row(LocalTime.of(23, 30, 0)) :: Nil)
+
+ // Exact bucket boundary
+ checkAnswer(
+ sql("SELECT time_bucket(INTERVAL '15' MINUTE, TIME'09:30:00')"),
+ Row(LocalTime.of(9, 30, 0)) :: Nil)
+ }
+
+ test("time_bucket with null inputs") {
+ checkAnswer(
+ sql("SELECT time_bucket(INTERVAL '15' MINUTE, NULL)"),
+ Row(null) :: Nil)
+
+ checkAnswer(
+ sql("SELECT time_bucket(NULL, TIME'12:34:56')"),
+ Row(null) :: Nil)
+
+ checkAnswer(
+ sql("SELECT time_bucket(NULL, NULL)"),
+ Row(null) :: Nil)
+ }
+
+ test("time_bucket with table data") {
+ withTable("time_bucket_test") {
+ sql("""
+ CREATE TABLE time_bucket_test (
+ id INT,
+ event_time TIME(6)
+ ) USING parquet
+ """)
+
+ sql("""
+ INSERT INTO time_bucket_test VALUES
+ (1, TIME'09:15:30'),
+ (2, TIME'09:37:45'),
+ (3, TIME'10:05:12'),
+ (4, TIME'14:42:00'),
+ (5, TIME'14:55:30')
+ """)
+
+ // Group by 30-minute buckets
+ checkAnswer(
+ sql("""
+ SELECT time_bucket(INTERVAL '30' MINUTE, event_time) AS time_slot,
+ COUNT(*) AS cnt
+ FROM time_bucket_test
+ GROUP BY time_slot
+ ORDER BY time_slot
+ """),
+ Row(LocalTime.of(9, 0, 0), 1L) ::
+ Row(LocalTime.of(9, 30, 0), 1L) ::
+ Row(LocalTime.of(10, 0, 0), 1L) ::
+ Row(LocalTime.of(14, 30, 0), 2L) :: Nil)
+ }
+ }
+
+ test("time_bucket error handling - zero interval") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SELECT time_bucket(INTERVAL '0' MINUTE,
TIME'12:34:56')").collect()
+ },
+ condition = "DATATYPE_MISMATCH.VALUE_OUT_OF_RANGE",
+ parameters = Map(
+ "sqlExpr" -> "\"time_bucket(INTERVAL '00' MINUTE, TIME '12:34:56')\"",
+ "exprName" -> "bucket_width",
+ "valueRange" -> "(0, 9223372036854775807]",
+ "currentValue" -> "0L"
+ ),
+ context = ExpectedContext(
+ fragment = "time_bucket(INTERVAL '0' MINUTE, TIME'12:34:56')",
+ start = 7,
+ stop = 54
+ )
+ )
+ }
+
+ test("time_bucket error handling - negative interval") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SELECT time_bucket(INTERVAL '-15' MINUTE,
TIME'12:34:56')").collect()
+ },
+ condition = "DATATYPE_MISMATCH.VALUE_OUT_OF_RANGE",
+ parameters = Map(
+ "sqlExpr" -> "\"time_bucket(INTERVAL '-15' MINUTE, TIME '12:34:56')\"",
+ "exprName" -> "bucket_width",
+ "valueRange" -> "(0, 9223372036854775807]",
+ "currentValue" -> "-900000000L"
+ ),
+ context = ExpectedContext(
+ fragment = "time_bucket(INTERVAL '-15' MINUTE, TIME'12:34:56')",
+ start = 7,
+ stop = 56
+ )
+ )
+ }
+
+ test("time_bucket DataFrame API") {
+ val df = Seq(
+ (1, "09:37:22"),
+ (2, "14:47:00"),
+ (3, "10:05:30")
+ ).toDF("id", "t")
+ .withColumn("t", to_time(col("t")))
+
+ // Using expr for interval
+ val result = df.withColumn(
+ "bucket",
+ time_bucket(expr("INTERVAL '30' MINUTE"), col("t"))
Review Comment:
Coverage gap: this "DataFrame API" test uses the `expr("INTERVAL '30'
MINUTE")` Column overload, so the `time_bucket(String, Column)` overload is
never exercised. There's also no test for a width ≥ 24h or a very large
interval (the degenerate `00:00:00` and overflow paths). Worth adding a
String-overload case and a ≥ 24h case once the width validation above is in.
--
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]
