vinodkc opened a new pull request, #53237:
URL: https://github.com/apache/spark/pull/53237
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR:
https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g.,
'[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a
faster review.
7. If you want to add a new configuration, please read the guideline first
for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the
guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section
is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster
reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class
hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other
DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
This PR adds a new `time_bucket()` SQL function that buckets TIME values
into fixed-width intervals, returning the start time of each bucket. This
enables histogram generation and time-of-day pattern analysis for TIME columns.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
The TIME type currently lacks a bucketing function for aggregation and
analysis. Users cannot easily group TIME values by arbitrary intervals (e.g.,
15-minute or 1-hour buckets) without complex manual calculations.
**Current Gap:**
Existing functions don't support TIME bucketing:
- `window()`: Only works with TIMESTAMP, not TIME. Returns a struct, not a
scalar.
- `date_trunc()`: Doesn't support TIME type
- `time_trunc()`: Only supports fixed calendar units (HOUR, MINUTE), not
arbitrary intervals like "15 minutes" or "90 minutes"
Current workarounds are error-prone, hard to maintain:
```sql
-- Manual calculation (error-prone, hard to maintain)
SELECT TIME(FLOOR(TIME_TO_SECONDS(event_time) / 900) * 900) as bucket FROM
events;
```
**Use Cases:**
This function addresses common real-world analytics needs:
1. Retail Analytics: Analyze customer traffic by 30-minute slots to optimize
staffing
2. Healthcare: Group appointments by 15-minute intervals for scheduling
optimization
3. Manufacturing: Aggregate sensor readings by hourly buckets to detect
production patterns
4. DevOps: Bucket system events by 5-minute intervals for performance
monitoring
5. Business Intelligence: Create time-of-day histograms for reporting
**Industry Precedent:**
- SQL Server 2022: `DATE_BUCKET()` supports TIME type bucketing
- TimescaleDB: `time_bucket()` is one of their most popular functions for
time-series analytics
- This fills a critical gap in Spark's TIME type functionality and brings it
on par with leading databases
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as
new features, bug fixes, or other behavior changes. Documentation-only updates
are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes
- provide the console output, description and/or an example to show the
behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to
the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
Yes. This PR adds a new SQL function `time_bucket()` available in SQL,
Scala, Python, and Spark Connect.
#### Function Signature
```sql
time_bucket(bucket_width, time) -> TIME
```
**Parameters:**
- `bucket_width`: A day-time interval expression (e.g., `INTERVAL '15'
MINUTE`)
- `time`: A TIME value to bucket
**Behavior:**
- Returns the start of the time bucket containing the input time
- Buckets are aligned to midnight (00:00:00)
- Buckets cannot span across midnight
- Returns the same precision as the input TIME type
- Returns NULL if either input is NULL
---
#### Examples
**Example 1: Basic Bucketing**
```sql
-- 15-minute buckets
SELECT time_bucket(INTERVAL '15' MINUTE, TIME'09:37:22');
-- Result: 09:30:00
-- 30-minute buckets
SELECT time_bucket(INTERVAL '30' MINUTE, TIME'14:47:00');
-- Result: 14:30:00
-- 1-hour buckets
SELECT time_bucket(INTERVAL '1' HOUR, TIME'16:35:00');
-- Result: 16:00:00
-- 2-hour buckets
SELECT time_bucket(INTERVAL '2' HOUR, TIME'15:20:00');
-- Result: 14:00:00
```
**Example 2: Retail Analytics - Peak Shopping Hours**
```sql
-- Find busiest 30-minute slots in a store
SELECT time_bucket(INTERVAL '30' MINUTE, purchase_time) AS time_slot,
COUNT(*) AS customer_count,
SUM(total_amount) AS revenue
FROM sales
WHERE date = '2024-01-15'
GROUP BY time_slot
ORDER BY customer_count DESC
LIMIT 10;
-- Sample Output:
-- +----------+---------------+---------+
-- |time_slot |customer_count |revenue |
-- +----------+---------------+---------+
-- |14:00:00 | 245 | 12450.50|
-- |14:30:00 | 231 | 11890.25|
-- |12:00:00 | 198 | 9875.00|
-- +----------+---------------+---------+
```
**Example 3: Healthcare - Appointment Scheduling**
```sql
-- Analyze appointment distribution by 15-minute slots
SELECT time_bucket(INTERVAL '15' MINUTE, appointment_time) AS slot,
COUNT(*) AS appointments,
AVG(duration_minutes) AS avg_duration,
SUM(CASE WHEN status = 'no_show' THEN 1 ELSE 0 END) AS no_shows
FROM appointments
WHERE appointment_date BETWEEN '2024-01-01' AND '2024-01-31'
GROUP BY slot
ORDER BY slot;
-- Sample Output:
-- +----------+-------------+-------------+---------+
-- |slot |appointments |avg_duration |no_shows |
-- +----------+-------------+-------------+---------+
-- |08:00:00 | 45 | 22.3 | 2 |
-- |08:15:00 | 48 | 24.1 | 3 |
-- |08:30:00 | 52 | 21.8 | 1 |
-- +----------+-------------+-------------+---------+
```
**Example 4: Edge Cases**
```sql
-- Midnight (start of day)
SELECT time_bucket(INTERVAL '1' HOUR, TIME'00:00:00');
-- Result: 00:00:00
-- Just before midnight
SELECT time_bucket(INTERVAL '1' HOUR, TIME'23:59:59.999999');
-- Result: 23:00:00
-- Microsecond precision
SELECT time_bucket(INTERVAL '1' MICROSECOND, TIME'12:34:56.123456');
-- Result: 12:34:56.123456
-- Millisecond buckets
SELECT time_bucket(INTERVAL '100' MILLISECOND, TIME'12:34:56.789123');
-- Result: 12:34:56.700000
-- Null handling
SELECT time_bucket(INTERVAL '15' MINUTE, NULL);
-- Result: NULL
SELECT time_bucket(NULL, TIME'12:34:56');
-- Result: NULL
```
---
#### Scala API
```scala
import org.apache.spark.sql.functions._
import java.time.LocalTime
val events = Seq(
(1, LocalTime.of(9, 5, 30), 45, 150.0),
(2, LocalTime.of(9, 37, 45), 67, 175.0),
(3, LocalTime.of(10, 12, 0), 28, 225.0)
).toDF("event_id", "event_time", "duration", "value")
events.createOrReplaceTempView("events")
val df = spark.table("events")
// Test Example 1
df.groupBy(time_bucket(expr("INTERVAL '15' MINUTE"),
col("event_time")).as("bucket"))
.agg(count("*").as("count"))
.orderBy("bucket")
.show()
// Test Example 2
df.groupBy(time_bucket("30 minutes", col("event_time")).as("bucket"))
.count()
.show()
// Test Example 3
df.groupBy(time_bucket("1 hour", col("event_time")).as("hour"))
.agg(
count("*").as("total_events"),
avg("duration").as("avg_duration"),
max("value").as("max_value")
)
.show()
```
#### Python API
```python
from pyspark.sql import functions as F
# Example 1: Basic bucketing
df = spark.table("events")
df.groupBy(F.time_bucket(F.expr("INTERVAL '15' MINUTE"),
"event_time").alias("bucket")) \
.count() \
.show()
# Example 2: Histogram generation
df.groupBy(F.time_bucket(F.expr("INTERVAL '30' MINUTE"),
"event_time").alias("slot")) \
.agg(
F.count("*").alias("count"),
F.avg("value").alias("avg_value"),
F.stddev("value").alias("stddev_value")
) \
.orderBy("slot") \
.show()
# Example 3: Peak detection
peak_hours = df.groupBy(
F.time_bucket(F.expr("INTERVAL '1' HOUR"), "purchase_time").alias("hour")
).agg(
F.sum("amount").alias("revenue")
).filter(
F.col("revenue") > 10000
).orderBy(F.desc("revenue"))
peak_hours.show()
```
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some
test cases that check the changes thoroughly including negative and positive
cases if possible.
If it was tested in a way different from regular unit tests, please clarify
how you tested step by step, ideally copy and paste-able, so that other
reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why
it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions
for the consistent environment, and the instructions could accord to:
https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Added tests in `TimeFunctionsSuiteBase` and `sql-tests/inputs/time.sql`
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this
patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling
Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No
--
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]
