risingwave_expr_impl/scalar/
date_bin.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use risingwave_common::types::{Interval, Timestamp, Timestamptz};
16use risingwave_expr::{ExprError, Result, function};
17
18#[function("date_bin(interval, timestamp, timestamp) -> timestamp")]
19pub fn date_bin_ts(stride: Interval, source: Timestamp, origin: Timestamp) -> Result<Timestamp> {
20    let source_us = source.0.and_utc().timestamp_micros(); // source to microseconds
21    let origin_us = origin.0.and_utc().timestamp_micros(); // origin to microseconds
22
23    let binned_source_us = date_bin_inner(stride, source_us, origin_us)?;
24    Timestamp::with_micros(binned_source_us).map_err(|_| ExprError::NumericOutOfRange)
25}
26
27#[function("date_bin(interval, timestamptz, timestamptz) -> timestamptz")]
28pub fn date_bin_tstz(
29    stride: Interval,
30    source: Timestamptz,
31    origin: Timestamptz,
32) -> Result<Timestamptz> {
33    let source_us = source.timestamp_micros(); // source to microseconds
34    let origin_us = origin.timestamp_micros(); // origin to microseconds
35
36    let binned_source_us = date_bin_inner(stride, source_us, origin_us)?;
37    Timestamptz::from_micros(binned_source_us).ok_or(ExprError::NumericOutOfRange)
38}
39
40fn date_bin_inner(stride: Interval, source_us: i64, origin_us: i64) -> Result<i64> {
41    if stride.months() != 0 {
42        // PostgreSQL doesn't allow months in the interval for date_bin.
43        return Err(ExprError::InvalidParam {
44            name: "stride",
45            reason: "stride interval with months not supported in date_bin".into(),
46        });
47    }
48    let stride_us = stride.usecs() + (stride.days() as i64) * Interval::USECS_PER_DAY; // stride width in microseconds
49
50    if stride_us <= 0 {
51        return Err(ExprError::InvalidParam {
52            name: "stride",
53            reason: "stride interval must be positive".into(),
54        });
55    }
56
57    // Compute how far ts is from the origin
58    let delta = source_us - origin_us;
59
60    // Floor the delta to the nearest stride
61    let bucket = delta.div_euclid(stride_us) * stride_us;
62
63    // Add back to origin to get stridened timestamp
64    let binned_source_us = origin_us + bucket;
65    Ok(binned_source_us)
66}
risingwave_expr_impl/scalar/date_bin.rs

risingwave_expr_impl/scalar/
date_bin.rs
