You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/23 16:46:36 UTC
(datafusion) 01/01: perf: add typed bounds for date_bin and from_unixtime (#25575)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25575-c9fcb2298f3e18e9a762750aedfb417106ae457e
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit fa9755bddcc2134f5f7fb5e01c5f60d67c8b8cdf
Author: Malingshu <69...@users.noreply.github.com>
AuthorDate: Wed Sep 23 15:11:49 2026 +0000
perf: add typed bounds for date_bin and from_unixtime (#25575)
## Which issue does this PR close?
- Closes #25465.
## Rationale for this change
`date_bin` and `from_unixtime` currently report an unbounded `Null`
interval from `evaluate_bounds`. This loses their output type during
physical property analysis, so safe timezone-free expressions can retain
redundant sorts.
## What changes are included in this PR?
- Return typed unbounded bounds from `date_bin` using its source type.
- Return typed unbounded bounds from `from_unixtime` when its session or
explicit timezone is known, while keeping value-dependent unknown
timezones unknown.
- Document when scalar UDFs should provide typed unbounded bounds.
- Add plan coverage for redundant sort removal and retain the
named-timezone correctness regression case.
## What is the testing strategy for this PR?
Unit tests cover timestamp, timezone-aware timestamp, `Time32`,
`Time64`, session timezone, constant explicit timezone, and unknown
timezone bounds. SQL logic tests cover redundant sort removal through
timezone-free `date_bin` and `from_unixtime`, plus named-timezone
correctness.
The following checks pass:
- `cargo fmt --all`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test -p datafusion-functions evaluate_bounds --lib`
- `cargo test --test sqllogictests -- timestamps.slt`
- The extended workspace test suite required by the contributor guide
## Are there any user-facing changes?
Queries that apply `date_trunc` to timezone-free `date_bin` or
`from_unixtime` results can avoid redundant sorts. Named-timezone cases
remain conservatively unordered.
---
datafusion/expr/src/udf.rs | 13 +++
datafusion/functions/src/datetime/date_bin.rs | 32 ++++++-
datafusion/functions/src/datetime/from_unixtime.rs | 105 ++++++++++++++++++---
.../test_files/datetime/timestamps.slt | 55 ++++++++++-
4 files changed, 192 insertions(+), 13 deletions(-)
diff --git a/datafusion/expr/src/udf.rs b/datafusion/expr/src/udf.rs
index 4340fbb4a3..3a79792eb6 100644
--- a/datafusion/expr/src/udf.rs
+++ b/datafusion/expr/src/udf.rs
@@ -904,6 +904,19 @@ pub trait ScalarUDFImpl: Debug + DynEq + DynHash + Send + Sync + Any {
///
/// If the function is `ABS(a)`, and the input interval is `a: [-3, 2]`,
/// then the output interval would be `[0, 3]`.
+ ///
+ /// # Output type
+ ///
+ /// The default implementation returns an unbounded [`DataType::Null`]
+ /// interval because the output type cannot be inferred generically.
+ /// Implementations should override this method when they can determine the
+ /// output type, even if they cannot compute precise bounds. Returning an
+ /// unbounded interval with the correct type preserves information used by
+ /// downstream physical property analysis.
+ ///
+ /// If the output type depends on argument values, only report it when those
+ /// values are known, such as from singleton input intervals. Otherwise,
+ /// keep the output type unknown.
fn evaluate_bounds(&self, _input: &[&Interval]) -> Result<Interval> {
// We cannot assume the input datatype is the same of output type.
Interval::make_unbounded(&DataType::Null)
diff --git a/datafusion/functions/src/datetime/date_bin.rs b/datafusion/functions/src/datetime/date_bin.rs
index 15cdecc3c2..909a7dbb75 100644
--- a/datafusion/functions/src/datetime/date_bin.rs
+++ b/datafusion/functions/src/datetime/date_bin.rs
@@ -38,6 +38,7 @@ use datafusion_common::{
Result, ScalarValue, exec_datafusion_err, exec_err, not_impl_err, plan_err,
};
use datafusion_expr::TypeSignature::Exact;
+use datafusion_expr::interval_arithmetic::Interval as ExprInterval;
use datafusion_expr::sort_properties::{ExprProperties, SortProperties};
use datafusion_expr::{
ColumnarValue, Documentation, ScalarFunctionArgs, ScalarUDFImpl, Signature,
@@ -283,6 +284,12 @@ impl ScalarUDFImpl for DateBinFunc {
Ok(SortProperties::Unordered)
}
}
+
+ fn evaluate_bounds(&self, inputs: &[&ExprInterval]) -> Result<ExprInterval> {
+ // DATE_BIN returns the same type as its source argument.
+ ExprInterval::make_unbounded(&inputs[1].data_type())
+ }
+
fn documentation(&self) -> Option<&Documentation> {
self.doc()
}
@@ -813,10 +820,11 @@ mod tests {
use arrow::array::types::TimestampNanosecondType;
use arrow::array::{Array, IntervalDayTimeArray, TimestampNanosecondArray};
use arrow::compute::kernels::cast_utils::string_to_timestamp_nanos;
- use arrow::datatypes::{DataType, Field, FieldRef, TimeUnit};
+ use arrow::datatypes::{DataType, Field, FieldRef, IntervalUnit, TimeUnit};
use arrow_buffer::{IntervalDayTime, IntervalMonthDayNano};
use datafusion_common::{DataFusionError, ScalarValue};
+ use datafusion_expr::interval_arithmetic::Interval as ExprInterval;
use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl};
use chrono::TimeDelta;
@@ -867,6 +875,28 @@ mod tests {
);
}
+ #[test]
+ fn evaluate_bounds_preserves_source_type() {
+ let stride =
+ ExprInterval::make_unbounded(&DataType::Interval(IntervalUnit::DayTime))
+ .unwrap();
+ let source_types = [
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())),
+ DataType::Time32(TimeUnit::Millisecond),
+ DataType::Time64(TimeUnit::Microsecond),
+ ];
+ let function = DateBinFunc::new();
+
+ for source_type in source_types {
+ let source = ExprInterval::make_unbounded(&source_type).unwrap();
+ let bounds = function.evaluate_bounds(&[&stride, &source]).unwrap();
+
+ assert!(bounds.is_unbounded());
+ assert_eq!(bounds.data_type(), source_type);
+ }
+ }
+
#[test]
fn test_date_bin() {
let return_field = &Arc::new(Field::new(
diff --git a/datafusion/functions/src/datetime/from_unixtime.rs b/datafusion/functions/src/datetime/from_unixtime.rs
index 8af4553f5c..53a170c5c8 100644
--- a/datafusion/functions/src/datetime/from_unixtime.rs
+++ b/datafusion/functions/src/datetime/from_unixtime.rs
@@ -28,6 +28,7 @@ use chrono::{DateTime, NaiveDateTime, Offset, TimeDelta, TimeZone};
use datafusion_common::config::ConfigOptions;
use datafusion_common::{Result, ScalarValue, exec_err, internal_err};
use datafusion_expr::TypeSignature::Exact;
+use datafusion_expr::interval_arithmetic::Interval;
use datafusion_expr::sort_properties::{ExprProperties, SortProperties};
use datafusion_expr::{
ColumnarValue, Documentation, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF,
@@ -110,6 +111,14 @@ impl FromUnixtimeFunc {
}
}
+fn timezone_from_scalar(value: &ScalarValue) -> Option<Arc<str>> {
+ value
+ .try_as_str()
+ .flatten()
+ .filter(|timezone| !timezone.is_empty())
+ .map(Arc::from)
+}
+
impl ScalarUDFImpl for FromUnixtimeFunc {
fn name(&self) -> &str {
"from_unixtime"
@@ -134,17 +143,9 @@ impl ScalarUDFImpl for FromUnixtimeFunc {
)
} else {
args.scalar_arguments[1]
- .and_then(|sv| {
- sv.try_as_str()
- .flatten()
- .filter(|s| !s.is_empty())
- .map(|tz| {
- Field::new(
- self.name(),
- Timestamp(Second, Some(Arc::from(tz.to_string()))),
- true,
- )
- })
+ .and_then(timezone_from_scalar)
+ .map(|timezone| {
+ Field::new(self.name(), Timestamp(Second, Some(timezone)), true)
})
.map(Arc::new)
.map_or_else(
@@ -219,6 +220,22 @@ impl ScalarUDFImpl for FromUnixtimeFunc {
Ok(true)
}
+ fn evaluate_bounds(&self, inputs: &[&Interval]) -> Result<Interval> {
+ let data_type = match inputs {
+ [_] => Timestamp(Second, self.timezone.clone()),
+ // Literal properties encode a constant timezone as a singleton interval.
+ [_, timezone] if timezone.lower() == timezone.upper() => {
+ match timezone_from_scalar(timezone.lower()) {
+ Some(timezone) => Timestamp(Second, Some(timezone)),
+ None => DataType::Null,
+ }
+ }
+ _ => DataType::Null,
+ };
+
+ Interval::make_unbounded(&data_type)
+ }
+
fn documentation(&self) -> Option<&Documentation> {
self.doc()
}
@@ -334,6 +351,7 @@ mod test {
use datafusion_common::ScalarValue;
use datafusion_common::ScalarValue::Int64;
use datafusion_common::config::ConfigOptions;
+ use datafusion_expr::interval_arithmetic::Interval;
use datafusion_expr::{
ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
};
@@ -346,6 +364,71 @@ mod test {
/// represent (`-262143-01-01T00:00:00`).
const MIN_UTC_SECONDS: i64 = -8334601228800;
+ #[test]
+ fn evaluate_bounds_preserves_output_type() {
+ let seconds = Interval::make_unbounded(&DataType::Int64).unwrap();
+
+ let function = FromUnixtimeFunc::default();
+ let bounds = function.evaluate_bounds(&[&seconds]).unwrap();
+ assert!(bounds.is_unbounded());
+ assert_eq!(bounds.data_type(), DataType::Timestamp(Second, None));
+
+ let mut options = ConfigOptions::default();
+ options.execution.time_zone = Some("America/Denver".to_string());
+ let function = FromUnixtimeFunc::new_with_config(&options);
+ let bounds = function.evaluate_bounds(&[&seconds]).unwrap();
+ assert_eq!(
+ bounds.data_type(),
+ DataType::Timestamp(Second, Some("America/Denver".into()))
+ );
+
+ let timezone = ScalarValue::Utf8(Some("+08:00".to_string()));
+ let timezone = Interval::try_new(timezone.clone(), timezone).unwrap();
+ let bounds = function.evaluate_bounds(&[&seconds, &timezone]).unwrap();
+ assert_eq!(
+ bounds.data_type(),
+ DataType::Timestamp(Second, Some("+08:00".into()))
+ );
+ }
+
+ #[test]
+ fn evaluate_bounds_keeps_unknown_timezone_unknown() {
+ let seconds = Interval::make_unbounded(&DataType::Int64).unwrap();
+ let function = FromUnixtimeFunc::default();
+
+ let unknown = Interval::make_unbounded(&DataType::Utf8).unwrap();
+ assert_eq!(
+ function
+ .evaluate_bounds(&[&seconds, &unknown])
+ .unwrap()
+ .data_type(),
+ DataType::Null
+ );
+
+ let varying = Interval::try_new(
+ ScalarValue::Utf8(Some("+08:00".to_string())),
+ ScalarValue::Utf8(Some("+09:00".to_string())),
+ )
+ .unwrap();
+ assert_eq!(
+ function
+ .evaluate_bounds(&[&seconds, &varying])
+ .unwrap()
+ .data_type(),
+ DataType::Null
+ );
+
+ let empty = ScalarValue::Utf8(Some(String::new()));
+ let empty = Interval::try_new(empty.clone(), empty).unwrap();
+ assert_eq!(
+ function
+ .evaluate_bounds(&[&seconds, &empty])
+ .unwrap()
+ .data_type(),
+ DataType::Null
+ );
+ }
+
/// Invoke the single argument form with `datafusion.execution.time_zone`
/// set to `timezone`.
fn from_unixtime_session_tz(
diff --git a/datafusion/sqllogictest/test_files/datetime/timestamps.slt b/datafusion/sqllogictest/test_files/datetime/timestamps.slt
index 05622c22da..6894590e62 100644
--- a/datafusion/sqllogictest/test_files/datetime/timestamps.slt
+++ b/datafusion/sqllogictest/test_files/datetime/timestamps.slt
@@ -1679,7 +1679,7 @@ ORDER BY date_trunc('hour', ts);
562129260 562125600
562129259 562129200
-# An unknown range from a child function may hide a timezone-aware timestamp
+# Typed bounds from a child function must retain named timezone metadata
query II
SELECT
c,
@@ -1702,6 +1702,59 @@ ORDER BY date_trunc('hour', from_unixtime(c, 'America/Goose_Bay'));
562129260 562125600
562129259 562129200
+# Typed bounds allow date_trunc to propagate ordering through timezone-free from_unixtime
+query TT
+EXPLAIN SELECT date_trunc('hour', from_unixtime(c)) AS truncated
+FROM (
+ SELECT column1 AS c
+ FROM (
+ VALUES
+ (7200::bigint),
+ (3600::bigint)
+ )
+ ORDER BY c
+ LIMIT 2
+)
+ORDER BY date_trunc('hour', from_unixtime(c));
+----
+logical_plan
+01)Sort: truncated ASC NULLS LAST
+02)--Projection: date_trunc(Utf8("hour"), from_unixtime(c)) AS truncated
+03)----Sort: c ASC NULLS LAST, fetch=2
+04)------Projection: column1 AS c
+05)--------Values: (Int64(7200)), (Int64(3600))
+physical_plan
+01)ProjectionExec: expr=[date_trunc(hour, from_unixtime(column1@0)) as truncated]
+02)--SortExec: TopK(fetch=2), expr=[column1@0 ASC NULLS LAST], preserve_partitioning=[false]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Typed bounds allow date_trunc to propagate ordering through timezone-free date_bin
+query TT
+EXPLAIN SELECT date_trunc('hour', date_bin(INTERVAL '1 minute', ts)) AS truncated
+FROM (
+ SELECT arrow_cast(column1, 'Timestamp(Second)') AS ts
+ FROM (
+ VALUES
+ (7200::bigint),
+ (3600::bigint)
+ )
+ ORDER BY ts
+ LIMIT 2
+)
+ORDER BY date_trunc('hour', date_bin(INTERVAL '1 minute', ts));
+----
+logical_plan
+01)Sort: truncated ASC NULLS LAST
+02)--Projection: date_trunc(Utf8("hour"), date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"), ts)) AS truncated
+03)----Sort: ts ASC NULLS LAST, fetch=2
+04)------Projection: CAST(column1 AS Timestamp(s)) AS ts
+05)--------Values: (Int64(7200)), (Int64(3600))
+physical_plan
+01)ProjectionExec: expr=[date_trunc(hour, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }, ts@0)) as truncated]
+02)--SortExec: TopK(fetch=2), expr=[ts@0 ASC NULLS LAST], preserve_partitioning=[false]
+03)----ProjectionExec: expr=[CAST(column1@0 AS Timestamp(s)) as ts]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
statement ok
CREATE TABLE timestamp_strings(ts varchar)
AS VALUES
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org