diff --git a/datafusion/functions/src/datetime/date_bin.rs b/datafusion/functions/src/datetime/date_bin.rs index d702af392c7c3..24e21e8f78d48 100644 --- a/datafusion/functions/src/datetime/date_bin.rs +++ b/datafusion/functions/src/datetime/date_bin.rs @@ -53,6 +53,8 @@ use chrono::{DateTime, Datelike, Duration, Months, TimeDelta, Utc}; Calculates time intervals and returns the start of the interval nearest to the specified timestamp. Use `date_bin` to downsample time series data by grouping rows into time-based "bins" or "windows" and applying an aggregate or selector function to each window. For example, if you "bin" or "window" data into 15 minute intervals, an input timestamp of `2023-01-01T18:18:18Z` will be updated to the start time of the 15 minute bin it is in: `2023-01-01T18:15:00Z`. + +For timestamp inputs, binning beyond the supported numeric or calendar range returns an error. A NULL source or interval still produces NULL. "#, syntax_example = "date_bin(interval, expression[, origin_timestamp])", sql_example = r#"```sql @@ -503,7 +505,7 @@ fn checked_scale_to_nanos(x: i64, scale: i64) -> Result { } } -// Per-row failures map to NULL, so use Option in the hot path. +// Return None when binning needs the wider arithmetic path. #[inline] fn scale_and_bin_to_nanos( value: i64, @@ -518,26 +520,25 @@ fn scale_and_bin_to_nanos( } // Per-row timestamp binning shared by scalar and array paths. -// Source-value failures become None, which callers map to NULL. #[inline] fn date_bin_timestamp_value( value: i64, origin: i64, stride: i64, stride_fn: BinFunctions, -) -> Option { +) -> Result { let scale = timestamp_scale::(); match scale_and_bin_to_nanos(value, scale, origin, stride, stride_fn.narrow) { - Some(binned) => Some(binned / scale), + Some(binned) => Ok(binned / scale), None => { date_bin_timestamp_value_wide(value, scale, origin, stride, stride_fn.wide) + .ok_or_else(|| exec_datafusion_err!("DATE_BIN timestamp out of range")) } } } // Slow path for values whose i64 nanosecond computation overflows. Binning in -// i128 means that only a result outside the source type's range becomes NULL, -// instead of any value outside the i64 nanosecond range. +// i128 avoids rejecting valid results outside the i64 nanosecond range. #[cold] #[inline(never)] fn date_bin_timestamp_value_wide( @@ -718,41 +719,45 @@ fn date_bin_impl( Ok(match array { ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(v, tz_opt)) => { ColumnarValue::Scalar(ScalarValue::TimestampNanosecond( - v.and_then(|x| { + v.map(|x| { date_bin_timestamp_value::( x, origin, stride, stride_fn, ) - }), + }) + .transpose()?, tz_opt.clone(), )) } ColumnarValue::Scalar(ScalarValue::TimestampMicrosecond(v, tz_opt)) => { ColumnarValue::Scalar(ScalarValue::TimestampMicrosecond( - v.and_then(|x| { + v.map(|x| { date_bin_timestamp_value::( x, origin, stride, stride_fn, ) - }), + }) + .transpose()?, tz_opt.clone(), )) } ColumnarValue::Scalar(ScalarValue::TimestampMillisecond(v, tz_opt)) => { ColumnarValue::Scalar(ScalarValue::TimestampMillisecond( - v.and_then(|x| { + v.map(|x| { date_bin_timestamp_value::( x, origin, stride, stride_fn, ) - }), + }) + .transpose()?, tz_opt.clone(), )) } ColumnarValue::Scalar(ScalarValue::TimestampSecond(v, tz_opt)) => { ColumnarValue::Scalar(ScalarValue::TimestampSecond( - v.and_then(|x| { + v.map(|x| { date_bin_timestamp_value::( x, origin, stride, stride_fn, ) - }), + }) + .transpose()?, tz_opt.clone(), )) } @@ -794,10 +799,9 @@ fn date_bin_impl( { let array = as_primitive_array::(array)?; - // Per-row errors become NULL, matching scalar behavior. - let result: PrimitiveArray = array.unary_opt(|val| { + let result: PrimitiveArray = array.try_unary(|val| { date_bin_timestamp_value::(val, origin, stride, stride_fn) - }); + })?; let array = result.with_timezone_opt(tz_opt.cloned()); Ok(ColumnarValue::Array(Arc::new(array))) @@ -912,8 +916,13 @@ mod tests { use std::sync::Arc; use crate::datetime::date_bin::{DateBinFunc, date_bin_nanos_interval}; - use arrow::array::types::TimestampNanosecondType; - use arrow::array::{Array, IntervalDayTimeArray, TimestampNanosecondArray}; + use arrow::array::types::{ + ArrowTimestampType, TimestampMicrosecondType, TimestampMillisecondType, + TimestampNanosecondType, TimestampSecondType, + }; + use arrow::array::{ + Array, IntervalDayTimeArray, PrimitiveArray, TimestampNanosecondArray, + }; use arrow::compute::kernels::cast_utils::string_to_timestamp_nanos; use arrow::datatypes::{DataType, Field, FieldRef, TimeUnit}; @@ -1487,13 +1496,11 @@ mod tests { )), ]; - let result = invoke_date_bin_with_args(args, 1, return_field); - assert!(result.is_ok()); - if let ColumnarValue::Scalar(ScalarValue::TimestampMillisecond(val, _)) = - result.unwrap() - { - assert!(val.is_none(), "Expected None for out of range operation"); - } + let error = invoke_date_bin_with_args(args, 1, return_field).unwrap_err(); + assert_eq!( + error.strip_backtrace(), + "Execution error: DATE_BIN timestamp out of range" + ); let args = vec![ ColumnarValue::Scalar(ScalarValue::new_interval_mdn(1637426858, 0, 0)), ColumnarValue::Scalar(ScalarValue::TimestampMillisecond( @@ -1506,20 +1513,18 @@ mod tests { )), ]; - let result = invoke_date_bin_with_args(args, 1, return_field); - assert!(result.is_ok()); - if let ColumnarValue::Scalar(ScalarValue::TimestampMillisecond(val, _)) = - result.unwrap() - { - assert!(val.is_none(), "Expected None for out of range operation"); - } + let error = invoke_date_bin_with_args(args, 1, return_field).unwrap_err(); + assert_eq!( + error.strip_backtrace(), + "Execution error: DATE_BIN timestamp out of range" + ); } #[test] fn test_date_bin_compute_distance_i64_min() { // Regression for #22215: date_bin_nanos_interval on a source near i64::MIN // previously panicked inside compute_distance with "attempt to subtract with overflow". - // Now it must return a normal Err that the scalar pipeline maps to NULL. + // It must return an execution error rather than panic. let result = date_bin_nanos_interval(3, i64::MIN, 0); assert!( result.is_err(), @@ -1536,18 +1541,11 @@ mod tests { ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(Some(i64::MIN), None)), ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(Some(0), None)), ]; - let result = invoke_date_bin_with_args(args, 1, return_field); - assert!(result.is_ok(), "expected Ok with NULL, got {result:?}"); - if let ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(val, _)) = - result.unwrap() - { - assert!( - val.is_none(), - "Expected None for compute_distance overflow, got {val:?}" - ); - } else { - panic!("Expected TimestampNanosecond scalar"); - } + let error = invoke_date_bin_with_args(args, 1, return_field).unwrap_err(); + assert_eq!( + error.strip_backtrace(), + "Execution error: DATE_BIN timestamp out of range" + ); } #[test] @@ -1610,7 +1608,7 @@ mod tests { } #[test] - fn test_date_bin_unrepresentable_result_returns_null() { + fn test_date_bin_unrepresentable_result_returns_error() { let epoch = || ScalarValue::TimestampNanosecond(Some(0), None); let cases = [ // The bin starts before the minimum value. @@ -1618,6 +1616,14 @@ mod tests { ScalarValue::new_interval_dt(1, 0), ScalarValue::TimestampSecond(Some(i64::MIN), None), ), + ( + ScalarValue::new_interval_dt(1, 0), + ScalarValue::TimestampMillisecond(Some(i64::MIN), None), + ), + ( + ScalarValue::new_interval_dt(1, 0), + ScalarValue::TimestampMicrosecond(Some(i64::MIN), None), + ), ( ScalarValue::new_interval_mdn(0, 0, 3), ScalarValue::TimestampNanosecond(Some(i64::MIN), None), @@ -1629,11 +1635,121 @@ mod tests { ), ]; for (stride, source) in cases { - let expected = ScalarValue::try_from(source.data_type()).unwrap(); - assert_date_bin(stride, source, epoch(), expected); + let return_field = Arc::new(Field::new("f", source.data_type(), true)); + for source in [ + ColumnarValue::Scalar(source.clone()), + ColumnarValue::Array(source.to_array().unwrap()), + ] { + let args = vec![ + ColumnarValue::Scalar(stride.clone()), + source, + ColumnarValue::Scalar(epoch()), + ]; + let error = invoke_date_bin_with_args(args, 1, &return_field) + .expect_err("out-of-range timestamp binning must fail"); + assert_eq!( + error.strip_backtrace(), + "Execution error: DATE_BIN timestamp out of range" + ); + } + } + } + + fn timestamp_binning_strides() -> [(&'static str, ScalarValue); 5] { + [ + ("three nanoseconds", ScalarValue::new_interval_mdn(0, 0, 3)), + ("one second", ScalarValue::new_interval_dt(0, 1000)), + ("one hour", ScalarValue::new_interval_dt(0, 3_600_000)), + ("one day", ScalarValue::new_interval_dt(1, 0)), + ("one month", ScalarValue::new_interval_mdn(1, 0, 0)), + ] + } + + fn assert_timestamp_nullness_preserved() { + let input = PrimitiveArray::::from_iter([Some(-1), None, Some(0), Some(1)]); + let return_field = Arc::new(Field::new("f", input.data_type().clone(), true)); + + for (stride_name, stride) in timestamp_binning_strides() { + for origin in [0, 1_234_567] { + let origin = ScalarValue::TimestampNanosecond(Some(origin), None); + let args = vec![ + ColumnarValue::Scalar(stride.clone()), + ColumnarValue::Array(Arc::new(input.clone())), + ColumnarValue::Scalar(origin.clone()), + ]; + let ColumnarValue::Array(output) = + invoke_date_bin_with_args(args, input.len(), &return_field).unwrap() + else { + panic!("expected array output"); + }; + assert_eq!( + input.nulls(), + output.nulls(), + "stride {stride_name}, origin {origin:?}, unit {:?}", + T::UNIT, + ); + } + } + } + + fn assert_timestamp_scalar_and_array_agree() { + let input = PrimitiveArray::::from_iter([Some(-1), None, Some(0), Some(1)]); + let return_field = Arc::new(Field::new("f", input.data_type().clone(), true)); + + for (stride_name, stride) in timestamp_binning_strides() { + for origin in [0, 1_234_567] { + let origin = ScalarValue::TimestampNanosecond(Some(origin), None); + let array_args = vec![ + ColumnarValue::Scalar(stride.clone()), + ColumnarValue::Array(Arc::new(input.clone())), + ColumnarValue::Scalar(origin.clone()), + ]; + let ColumnarValue::Array(array_output) = + invoke_date_bin_with_args(array_args, input.len(), &return_field) + .unwrap() + else { + panic!("expected array output"); + }; + + for index in 0..input.len() { + let source = ScalarValue::try_from_array(&input, index).unwrap(); + let scalar_args = vec![ + ColumnarValue::Scalar(stride.clone()), + ColumnarValue::Scalar(source), + ColumnarValue::Scalar(origin.clone()), + ]; + let ColumnarValue::Scalar(scalar_output) = + invoke_date_bin_with_args(scalar_args, 1, &return_field).unwrap() + else { + panic!("expected scalar output"); + }; + assert_eq!( + scalar_output, + ScalarValue::try_from_array(&array_output, index).unwrap(), + "stride {stride_name}, origin {origin:?}, unit {:?}, index {index}", + T::UNIT, + ); + } + } } } + #[test] + fn test_date_bin_preserves_timestamp_nullness() { + assert_timestamp_nullness_preserved::(); + assert_timestamp_nullness_preserved::(); + assert_timestamp_nullness_preserved::(); + assert_timestamp_nullness_preserved::(); + } + + #[test] + fn test_date_bin_timestamp_scalar_and_array_agree() { + assert_timestamp_scalar_and_array_agree::(); + assert_timestamp_scalar_and_array_agree::(); + assert_timestamp_scalar_and_array_agree::(); + assert_timestamp_scalar_and_array_agree::(); + } + #[test] fn test_date_bin_time64_micro_overflow_handling() { // Time64(Microsecond) can hold out-of-range values after reinterpret casts. @@ -1670,7 +1786,7 @@ mod tests { assert_overflow_error(invoke_date_bin_with_args(args, 2, return_field)); } - // Compare scalar execution with a one-row array for the same input. + // Compare values or errors from scalar execution and a one-row array. fn assert_scalar_array_parity( stride: ScalarValue, source: ScalarValue, @@ -1683,11 +1799,7 @@ mod tests { ColumnarValue::Scalar(source.clone()), ColumnarValue::Scalar(origin.clone()), ]; - let scalar_result = invoke_date_bin_with_args(scalar_args, 1, &return_field) - .expect("scalar path should not error"); - let ColumnarValue::Scalar(scalar_value) = scalar_result else { - panic!("expected scalar result, got {scalar_result:?}"); - }; + let scalar_result = invoke_date_bin_with_args(scalar_args, 1, &return_field); let source_array = source.to_array().expect("source value to array"); let array_args = vec![ @@ -1695,18 +1807,34 @@ mod tests { ColumnarValue::Array(source_array), ColumnarValue::Scalar(origin), ]; - let array_result = invoke_date_bin_with_args(array_args, 1, &return_field) - .expect("array path should not error"); - let ColumnarValue::Array(array) = array_result else { - panic!("expected array result, got {array_result:?}"); - }; - let array_value = - ScalarValue::try_from_array(&array, 0).expect("array row to scalar"); + let array_result = invoke_date_bin_with_args(array_args, 1, &return_field); - assert_eq!( - scalar_value, array_value, - "scalar and array results diverged for source {source:?}" - ); + match (scalar_result, array_result) { + ( + Ok(ColumnarValue::Scalar(scalar_value)), + Ok(ColumnarValue::Array(array)), + ) => { + let array_value = + ScalarValue::try_from_array(&array, 0).expect("array row to scalar"); + assert_eq!( + scalar_value, array_value, + "scalar and array values diverged for source {source:?}" + ); + } + (Err(scalar_error), Err(array_error)) => { + assert_eq!( + scalar_error.strip_backtrace(), + array_error.strip_backtrace(), + "scalar and array errors diverged for source {source:?}" + ); + } + (scalar_result, array_result) => { + panic!( + "scalar and array outcomes diverged for source {source:?}: \ + scalar {scalar_result:?}, array {array_result:?}" + ); + } + } } #[test] @@ -1720,14 +1848,14 @@ mod tests { ScalarValue::TimestampNanosecond(Some(0), None), ); - // Source scaling overflow -> NULL in both paths. + // Coarse timestamps use wide arithmetic when nanosecond scaling overflows. assert_scalar_array_parity( ScalarValue::new_interval_dt(1, 0), ScalarValue::TimestampSecond(Some(i64::MAX), None), ScalarValue::TimestampNanosecond(Some(0), None), ); - // Month interval out-of-range binning -> NULL in both paths. + // Month interval out-of-range binning must return the same error. assert_scalar_array_parity( ScalarValue::new_interval_mdn(1637426858, 0, 0), ScalarValue::TimestampMillisecond(Some(1040292460), None), diff --git a/datafusion/sqllogictest/test_files/date_bin_errors.slt b/datafusion/sqllogictest/test_files/date_bin_errors.slt index 68536ee455e4d..5ed88e062f405 100644 --- a/datafusion/sqllogictest/test_files/date_bin_errors.slt +++ b/datafusion/sqllogictest/test_files/date_bin_errors.slt @@ -17,11 +17,9 @@ # Tests for DATE_BIN error handling with out-of-range values -# Test case from issue #20219 - should return NULL instead of panicking -query P +# Issue #20219: out-of-range calendar arithmetic must error instead of panicking. +query error DataFusion error: Execution error: DATE_BIN timestamp out of range select date_bin(interval '1637426858 months', to_timestamp_millis(1040292460), timestamp '1984-01-07 00:00:00'); ----- -NULL # Issue #22528: negative sub-second source with month interval. query P @@ -36,42 +34,32 @@ from values (to_timestamp_millis(-1040292460)) t(c); ---- 1969-12-07T00:00:00 -# Array path should return NULL for per-row overflow. -query P +# Column inputs propagate the same error. +query error DataFusion error: Execution error: DATE_BIN timestamp out of range select date_bin(interval '1637426858 months', c, timestamp '1984-01-07 00:00:00') from values (to_timestamp_millis(1040292460)) t(c); ----- -NULL -# Large stride causing overflow - should return NULL -query P +# A large month stride can exceed the supported calendar range. +query error DataFusion error: Execution error: DATE_BIN timestamp out of range select date_bin( interval '1637426858 months', timestamp '1969-12-31 00:00:00', timestamp '1984-01-07 00:00:00' ); ----- -NULL -# Another large stride test -query P +query error DataFusion error: Execution error: DATE_BIN timestamp out of range select date_bin( interval '1637426858 months', to_timestamp_millis(-1040292000), timestamp '1984-01-07 00:00:00' ) as b; ----- -NULL -# Test with 1900-01-01 timestamp -query P +query error DataFusion error: Execution error: DATE_BIN timestamp out of range select date_bin( interval '1637426858 months', to_timestamp_millis(-2208988800000), timestamp '1984-01-07 00:00:00' ) as b; ----- -NULL # source - origin overflows i64 nanoseconds query P @@ -83,13 +71,48 @@ select date_bin( ---- 2262-04-11T23:47:16.854775807 -# compute_distance overflow: source at i64::MIN nanoseconds previously panicked -# inside compute_distance; it must return NULL through the SQL execution path -query P +# The bin starts below the minimum nanosecond timestamp. +query error DataFusion error: Execution error: DATE_BIN timestamp out of range select date_bin( interval '3 nanoseconds', arrow_cast(-9223372036854775808, 'Timestamp(Nanosecond, None)') ); + +# The same underflow must fail for column inputs, even alongside valid and NULL rows. +query error DataFusion error: Execution error: DATE_BIN timestamp out of range +select date_bin(interval '3 nanoseconds', arrow_cast(ts, 'Timestamp(Nanosecond, None)')) +from (values (0::bigint), (NULL::bigint), (-9223372036854775808)) t(ts); + +# Fixed-duration bins can underflow every timestamp precision. +query error DataFusion error: Execution error: DATE_BIN timestamp out of range +select date_bin(interval '1 day', arrow_cast(-9223372036854775808, 'Timestamp(Second, None)')); + +query error DataFusion error: Execution error: DATE_BIN timestamp out of range +select date_bin(interval '1 day', arrow_cast(-9223372036854775808, 'Timestamp(Millisecond, None)')); + +query error DataFusion error: Execution error: DATE_BIN timestamp out of range +select date_bin(interval '1 day', arrow_cast(-9223372036854775808, 'Timestamp(Microsecond, None)')); + +# A month calculation can exceed chrono's range even when the input type accepts it. +query error DataFusion error: Execution error: DATE_BIN timestamp out of range +select date_bin(interval '1 month', arrow_cast(9223372036854775807, 'Timestamp(Second, None)')); + +# NULL source values remain NULL, including in mixed column inputs. +query PPPP +select + date_bin(interval '1 day', arrow_cast(ts, 'Timestamp(Second, None)')), + date_bin(interval '1 day', arrow_cast(ts, 'Timestamp(Millisecond, None)')), + date_bin(interval '1 day', arrow_cast(ts, 'Timestamp(Microsecond, None)')), + date_bin(interval '1 day', arrow_cast(ts, 'Timestamp(Nanosecond, None)')) +from (values (0::bigint), (NULL::bigint)) t(ts) +order by ts NULLS LAST; +---- +1970-01-01T00:00:00 1970-01-01T00:00:00 1970-01-01T00:00:00 1970-01-01T00:00:00 +NULL NULL NULL NULL + +# A NULL interval still produces NULL for a non-null timestamp. +query P +select date_bin(NULL, timestamp '2026-01-01 00:00:00'); ---- NULL diff --git a/docs/source/user-guide/sql/scalar_functions.md b/docs/source/user-guide/sql/scalar_functions.md index 8b7cfffe1bb7e..05c7a35e0f44e 100644 --- a/docs/source/user-guide/sql/scalar_functions.md +++ b/docs/source/user-guide/sql/scalar_functions.md @@ -2473,6 +2473,8 @@ Calculates time intervals and returns the start of the interval nearest to the s For example, if you "bin" or "window" data into 15 minute intervals, an input timestamp of `2023-01-01T18:18:18Z` will be updated to the start time of the 15 minute bin it is in: `2023-01-01T18:15:00Z`. +For timestamp inputs, binning beyond the supported numeric or calendar range returns an error. A NULL source or interval still produces NULL. + ```sql date_bin(interval, expression[, origin_timestamp]) ```