viirya commented on code in PR #25668:
URL: https://github.com/apache/datafusion/pull/25668#discussion_r4147165788
##########
datafusion/physical-expr/src/scalar_function.rs:
##########
@@ -377,18 +386,234 @@ mod tests {
}
fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
- Ok(DataType::Int32)
+ Ok(self.return_type.clone())
}
fn invoke_with_args(&self, _args: ScalarFunctionArgs) ->
Result<ColumnarValue> {
- Ok(ColumnarValue::Scalar(ScalarValue::Int32(Some(42))))
+ Ok(ColumnarValue::Scalar(ScalarValue::try_from(
+ &self.return_type,
+ )?))
+ }
+ }
+
+ fn default_bounds_expr(return_type: DataType) -> ScalarFunctionExpr {
+ ScalarFunctionExpr::try_new(
+ Arc::new(ScalarUDF::from(MockScalarUDF {
+ signature: Signature::exact(vec![], Volatility::Immutable),
+ return_type,
+ })),
+ vec![],
+ &Schema::empty(),
+ Arc::new(ConfigOptions::default()),
+ )
+ .unwrap()
+ }
+
+ #[test]
+ fn properties_recover_return_type_without_changing_bounds_evaluation() {
+ for data_type in [
+ DataType::Null,
+ DataType::Boolean,
+ DataType::Int32,
+ DataType::UInt64,
+ DataType::Float64,
+ DataType::Utf8,
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Nanosecond,
Some("America/Goose_Bay".into())),
+ DataType::Time32(TimeUnit::Millisecond),
+ DataType::Time64(TimeUnit::Microsecond),
+ DataType::List(Arc::new(Field::new("item", DataType::Int32,
true))),
+ ] {
+ let expr = default_bounds_expr(data_type.clone());
+ let properties = expr.get_properties(&[]).unwrap();
+ assert_eq!(
+ properties.range,
+ Interval::make_unbounded(&data_type).unwrap()
+ );
+ assert_eq!(properties.sort_properties, SortProperties::Unordered);
+ assert!(!properties.preserves_lex_ordering);
+ assert!(!properties.strictly_order_preserving);
+ // The constraint solver calls this method directly, bypassing the
fallback.
Review Comment:
Same wording nit as in `udf.rs`: the cp_solver doesn't accept
`ScalarFunctionExpr`, so something like "Direct `evaluate_bounds` calls bypass
the fallback" would be more precise.
##########
datafusion/sqllogictest/test_files/datetime/arith_time_interval.slt:
##########
@@ -158,3 +158,58 @@ NULL
statement ok
DROP TABLE time_vals
+
+# Time arithmetic can wrap even when its input is an ordered function result.
Review Comment:
I don't think these two queries can catch a regression in the fallback.
SQL-planned `BinaryExpr`s have `fail_on_overflow = false`, and a function
output's range is unbounded, so `arithmetic_sort_properties` returns
`Unordered` through the overflow rule whether the left range is `Time64` or
`Null`. The sort stays either way, including with the fallback line removed.
The `wraps_in_domain` check, which is the part that actually depends on the
recovered type, is only reached with `with_fail_on_overflow(true)`. Could you
add a unit test that builds `date_trunc('hour', t) + INTERVAL '2 hours'` as a
`BinaryExpr` with `with_fail_on_overflow(true)` over an ordered `Time64` child
and asserts `Unordered`? The SLTs are still fine to keep as end-to-end result
checks.
##########
datafusion/physical-expr/src/scalar_function.rs:
##########
@@ -377,18 +386,234 @@ mod tests {
}
fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
- Ok(DataType::Int32)
+ Ok(self.return_type.clone())
}
fn invoke_with_args(&self, _args: ScalarFunctionArgs) ->
Result<ColumnarValue> {
- Ok(ColumnarValue::Scalar(ScalarValue::Int32(Some(42))))
+ Ok(ColumnarValue::Scalar(ScalarValue::try_from(
+ &self.return_type,
+ )?))
+ }
+ }
+
+ fn default_bounds_expr(return_type: DataType) -> ScalarFunctionExpr {
+ ScalarFunctionExpr::try_new(
+ Arc::new(ScalarUDF::from(MockScalarUDF {
+ signature: Signature::exact(vec![], Volatility::Immutable),
+ return_type,
+ })),
+ vec![],
+ &Schema::empty(),
+ Arc::new(ConfigOptions::default()),
+ )
+ .unwrap()
+ }
+
+ #[test]
+ fn properties_recover_return_type_without_changing_bounds_evaluation() {
+ for data_type in [
+ DataType::Null,
+ DataType::Boolean,
+ DataType::Int32,
+ DataType::UInt64,
+ DataType::Float64,
+ DataType::Utf8,
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Nanosecond,
Some("America/Goose_Bay".into())),
+ DataType::Time32(TimeUnit::Millisecond),
+ DataType::Time64(TimeUnit::Microsecond),
+ DataType::List(Arc::new(Field::new("item", DataType::Int32,
true))),
+ ] {
+ let expr = default_bounds_expr(data_type.clone());
+ let properties = expr.get_properties(&[]).unwrap();
+ assert_eq!(
+ properties.range,
+ Interval::make_unbounded(&data_type).unwrap()
+ );
+ assert_eq!(properties.sort_properties, SortProperties::Unordered);
+ assert!(!properties.preserves_lex_ordering);
+ assert!(!properties.strictly_order_preserving);
+ // The constraint solver calls this method directly, bypassing the
fallback.
+ assert_eq!(
+ expr.evaluate_bounds(&[]).unwrap(),
+ Interval::make_unbounded(&DataType::Null).unwrap()
+ );
+ }
+ }
+
+ #[test]
+ fn properties_keep_unknown_bounds_for_unsupported_return_type() {
+ // This unit is not supported for Time32 by ScalarValue/Interval.
+ let data_type = DataType::Time32(TimeUnit::Nanosecond);
+ assert!(Interval::make_unbounded(&data_type).is_err());
+ let properties =
default_bounds_expr(data_type).get_properties(&[]).unwrap();
+ assert_eq!(
+ properties.range,
+ Interval::make_unbounded(&DataType::Null).unwrap()
+ );
+ }
+
+ #[test]
+ fn datetime_properties_use_resolved_return_types() {
+ use datafusion_functions::datetime::date_bin::DateBinFunc;
+ use datafusion_functions::datetime::from_unixtime::FromUnixtimeFunc;
+
+ for source_type in [
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())),
+ DataType::Time32(TimeUnit::Millisecond),
+ DataType::Time64(TimeUnit::Microsecond),
+ ] {
+ let expr = ScalarFunctionExpr::try_new(
+ Arc::new(ScalarUDF::from(DateBinFunc::new())),
+ vec![
+ Arc::new(Literal::new(ScalarValue::new_interval_mdn(
+ 0,
+ 0,
+ 60_000_000_000,
+ ))),
+ Arc::new(Column::new("ts", 0)),
+ ],
+ &Schema::new(vec![Field::new("ts", source_type.clone(),
true)]),
+ Arc::new(ConfigOptions::default()),
+ )
+ .unwrap();
+ let properties = expr
+ .get_properties(&[
+
ExprProperties::new_unknown().with_order(SortProperties::Singleton),
+ ExprProperties::new_unknown()
+
.with_range(Interval::make_unbounded(&source_type).unwrap()),
+ ])
+ .unwrap();
+ assert_eq!(
+ properties.range,
+ Interval::make_unbounded(&source_type).unwrap()
+ );
+ }
+
+ for (session_tz, explicit_tz, expected_tz) in [
+ (None, None, None),
+ (Some("America/Denver"), None, Some("America/Denver")),
+ (
+ Some("America/Denver"),
+ Some("America/Goose_Bay"),
+ Some("America/Goose_Bay"),
+ ),
+ ] {
+ let mut config = ConfigOptions::default();
+ config.execution.time_zone = session_tz.map(str::to_owned);
+ let udf = FromUnixtimeFunc::new_with_config(&config);
+ let mut args: Vec<Arc<dyn PhysicalExpr>> =
+ vec![Arc::new(Column::new("c", 0))];
+ if let Some(tz) = explicit_tz {
+ args.push(Arc::new(Literal::new(ScalarValue::from(tz))));
+ }
+ let children = vec![ExprProperties::new_unknown(); args.len()];
+ let expr = ScalarFunctionExpr::try_new(
+ Arc::new(ScalarUDF::from(udf)),
+ args,
+ &Schema::new(vec![Field::new("c", DataType::Int64, true)]),
+ Arc::new(config),
+ )
+ .unwrap();
+ // The literal's timezone was resolved during construction, even
+ // though the child intervals here contain no value information.
+ let expected =
+ DataType::Timestamp(TimeUnit::Second,
expected_tz.map(Into::into));
+ assert_eq!(
+ expr.get_properties(&children).unwrap().range,
+ Interval::make_unbounded(&expected).unwrap()
+ );
+ }
+ }
+
+ #[derive(Debug, PartialEq, Eq, Hash)]
+ struct BoundsUDF {
+ inner: MockScalarUDF,
+ fail: bool,
+ }
+
+ impl ScalarUDFImpl for BoundsUDF {
+ fn name(&self) -> &str {
+ "bounds_function"
+ }
+
+ fn signature(&self) -> &Signature {
+ self.inner.signature()
+ }
+
+ fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
+ self.inner.return_type(arg_types)
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) ->
Result<ColumnarValue> {
+ self.inner.invoke_with_args(args)
+ }
+
+ fn evaluate_bounds(&self, inputs: &[&Interval]) -> Result<Interval> {
+ if self.fail {
+ return internal_err!("bounds evaluation failed");
+ }
+ Ok(inputs[0].clone())
+ }
+ }
+
+ #[test]
+ fn properties_preserve_udf_bounds_and_errors() {
+ for fail in [false, true] {
Review Comment:
nit: the `fail` / non-`fail` branches share little besides setup; two
separate tests (bounds preserved, error propagated) would read more easily than
the loop plus `if fail`.
##########
datafusion/expr/src/udf.rs:
##########
@@ -908,15 +908,15 @@ pub trait ScalarUDFImpl: Debug + DynEq + DynHash + Send +
Sync + Any {
/// # 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.
+ /// interval because the output type cannot be inferred from input
intervals
+ /// alone. During physical property analysis, `ScalarFunctionExpr` uses its
+ /// resolved return type to replace this default with a typed unbounded
+ /// interval when supported. UDFs need not override this method solely to
+ /// preserve their output type for that analysis.
+ ///
+ /// Override this method to provide more precise bounds. Explicit bounds
+ /// are preserved. The fallback does not apply to direct bounds evaluation
Review Comment:
Two small wording suggestions:
- "direct bounds evaluation by the constraint solver": `check_support` in
`physical-expr/src/intervals/utils.rs` doesn't accept `ScalarFunctionExpr`, so
the cp_solver never reaches this path. Maybe "direct calls to
`PhysicalExpr::evaluate_bounds`" is more accurate.
- "Explicit bounds are preserved": it may help to spell out the exact rule,
i.e. *any* unbounded `Null` result is replaced (including one an override
returns on purpose), and any other interval is used as-is, even if its type
differs from the resolved return type.
--
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]
