From d018d5fb144705cc74978f5666ec552a86a20298 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Fri, 15 May 2026 19:47:40 +0200 Subject: [PATCH 01/14] feat: extend check_support to include ScalarFunctionExpr in interval analysis `check_support` previously returned `false` for any `ScalarFunctionExpr`, preventing `FilterExec::statistics()` from entering the interval-analysis path for predicates containing scalar UDFs (e.g. `ceil(x) > 100`). Extend `check_support` to recurse into `ScalarFunctionExpr` when both the return type and all argument expressions are supported. This enables `evaluate_bounds` and `propagate_constraints` on `ScalarUDFImpl` to participate in selectivity estimation and constraint propagation. --- .../physical-expr/src/intervals/utils.rs | 113 ++++++++++++++++++ 1 file changed, 113 insertions(+) diff --git a/datafusion/physical-expr/src/intervals/utils.rs b/datafusion/physical-expr/src/intervals/utils.rs index 1090660a6b5e6..b62b6c260f5c7 100644 --- a/datafusion/physical-expr/src/intervals/utils.rs +++ b/datafusion/physical-expr/src/intervals/utils.rs @@ -22,6 +22,7 @@ use std::sync::Arc; use crate::{ PhysicalExpr, expressions::{BinaryExpr, CastExpr, Column, Literal, NegativeExpr}, + scalar_function::ScalarFunctionExpr, }; use arrow::array::types::{IntervalDayTime, IntervalMonthDayNano}; @@ -56,6 +57,12 @@ pub fn check_support(expr: &Arc, schema: &SchemaRef) -> bool { check_support(cast.expr(), schema) } else if let Some(negative) = expr.downcast_ref::() { check_support(negative.arg(), schema) + } else if let Some(scalar_fn) = expr.downcast_ref::() { + is_datatype_supported(scalar_fn.return_type()) + && scalar_fn + .args() + .iter() + .all(|arg| check_support(arg, schema)) } else { false } @@ -193,3 +200,109 @@ fn interval_dt_to_duration_ms(dt: &IntervalDayTime) -> Result { ) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::expressions::{Column, Literal}; + use crate::scalar_function::ScalarFunctionExpr; + use arrow::datatypes::{Field, Schema}; + use datafusion_common::ScalarValue; + use datafusion_common::config::ConfigOptions; + + fn f64_schema() -> SchemaRef { + Arc::new(Schema::new(vec![Field::new("x", DataType::Float64, false)])) + } + + fn utf8_schema() -> SchemaRef { + Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, false)])) + } + + fn col_x() -> Arc { + Arc::new(Column::new("x", 0)) + } + + fn lit_f64(v: f64) -> Arc { + Arc::new(Literal::new(ScalarValue::Float64(Some(v)))) + } + + fn scalar_fn_expr( + udf: Arc, + args: Vec>, + return_type: DataType, + ) -> Arc { + let name = udf.name().to_string(); + Arc::new(ScalarFunctionExpr::new( + &name, + udf, + args, + Field::new("result", return_type, true).into(), + Arc::new(ConfigOptions::default()), + )) + } + + #[test] + fn test_check_support_scalar_fn_supported_return_type() { + // ceil(x) returns Float64 — both return type and child are supported + let schema = f64_schema(); + let expr = scalar_fn_expr( + datafusion_functions::math::ceil(), + vec![col_x()], + DataType::Float64, + ); + assert!(check_support(&expr, &schema)); + } + + #[test] + fn test_check_support_scalar_fn_unsupported_return_type() { + // A UDF that returns Utf8 — not in is_datatype_supported + let schema = f64_schema(); + let expr = scalar_fn_expr( + datafusion_functions::math::ceil(), + vec![col_x()], + DataType::Utf8, + ); + assert!(!check_support(&expr, &schema)); + } + + #[test] + fn test_check_support_scalar_fn_unsupported_child() { + // ceil applied to a Utf8 column — child fails is_datatype_supported + let schema = utf8_schema(); + let col_s = Arc::new(Column::new("s", 0)) as Arc; + let expr = scalar_fn_expr( + datafusion_functions::math::ceil(), + vec![col_s], + DataType::Float64, + ); + assert!(!check_support(&expr, &schema)); + } + + #[test] + fn test_check_support_scalar_fn_in_binary_expr() { + // ceil(x) > 5.0 — the main use case: ScalarFunctionExpr inside a BinaryExpr + let schema = f64_schema(); + let ceil_x = scalar_fn_expr( + datafusion_functions::math::ceil(), + vec![col_x()], + DataType::Float64, + ); + let expr: Arc = + Arc::new(BinaryExpr::new(ceil_x, Operator::Gt, lit_f64(5.0))); + assert!(check_support(&expr, &schema)); + } + + #[test] + fn test_check_support_scalar_fn_in_binary_expr_unsupported_return() { + // f(x) > 5.0 where f returns Utf8 — should be false + let schema = f64_schema(); + let fn_expr = scalar_fn_expr( + datafusion_functions::math::ceil(), + vec![col_x()], + DataType::Utf8, + ); + let expr: Arc = + Arc::new(BinaryExpr::new(fn_expr, Operator::Gt, lit_f64(5.0))); + assert!(!check_support(&expr, &schema)); + } +} From 361e1658683e59242821a5d25ba5c062deb47f78 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sat, 16 May 2026 11:58:50 +0200 Subject: [PATCH 02/14] ceil and test --- datafusion/functions/src/math/ceil.rs | 59 ++++++++++++++++++++++++- datafusion/physical-plan/src/filter.rs | 60 ++++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 2 deletions(-) diff --git a/datafusion/functions/src/math/ceil.rs b/datafusion/functions/src/math/ceil.rs index 395cb4eae03f5..b67699a8bd7d9 100644 --- a/datafusion/functions/src/math/ceil.rs +++ b/datafusion/functions/src/math/ceil.rs @@ -191,11 +191,66 @@ impl ScalarUDFImpl for CeilFunc { } fn evaluate_bounds(&self, inputs: &[&Interval]) -> Result { - let data_type = inputs[0].data_type(); - Interval::make_unbounded(&data_type) + let [input] = inputs else { + return Interval::make_unbounded(&DataType::Float64); + }; + let data_type = input.data_type(); + match (ceil_scalar(input.lower()), ceil_scalar(input.upper())) { + (Some(lo), Some(hi)) => Interval::try_new(lo, hi) + .or_else(|_| Interval::make_unbounded(&data_type)), + _ => Interval::make_unbounded(&data_type), + } + } + + fn propagate_constraints( + &self, + interval: &Interval, + inputs: &[&Interval], + ) -> Result>> { + let [input_interval] = inputs else { + return Ok(Some(vec![])); + }; + // ceil(x) ∈ [N, M] → x ∈ (N−1, M] — conservative closed: [N−1, M] + let lo = match interval.lower() { + ScalarValue::Float64(Some(n)) if n.is_finite() => { + Some(ScalarValue::Float64(Some(n - 1.0))) + } + ScalarValue::Float32(Some(n)) if n.is_finite() => { + Some(ScalarValue::Float32(Some(n - 1.0))) + } + _ => None, + }; + let hi = match interval.upper() { + ScalarValue::Float64(Some(n)) if n.is_finite() => { + Some(ScalarValue::Float64(Some(*n))) + } + ScalarValue::Float32(Some(n)) if n.is_finite() => { + Some(ScalarValue::Float32(Some(*n))) + } + _ => None, + }; + match (lo, hi) { + (Some(lo), Some(hi)) => { + let constraint = Interval::try_new(lo, hi)?; + Ok(input_interval.intersect(constraint)?.map(|r| vec![r])) + } + _ => Ok(Some(vec![])), + } } fn documentation(&self) -> Option<&Documentation> { self.doc() } } + +fn ceil_scalar(v: &ScalarValue) -> Option { + match v { + ScalarValue::Float64(Some(f)) if f.is_finite() => { + Some(ScalarValue::Float64(Some(f.ceil()))) + } + ScalarValue::Float32(Some(f)) if f.is_finite() => { + Some(ScalarValue::Float32(Some(f.ceil()))) + } + _ => None, + } +} diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index c485e181f3826..c39f05d05ab77 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -1209,6 +1209,66 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_filter_statistics_ceil_scalar_fn() -> Result<()> { + // Table: x Float64, min=8.0, max=16.0, 100 rows. + // Filter: ceil(x) > 12.0 + // + // The range [8.0, 16.0] lies within a single IEEE-754 binade so + // Float64 cardinality is proportional to the value range, making + // the selectivity estimate predictable. + // + // With check_support recognising ScalarFunctionExpr and CeilFunc + // implementing evaluate_bounds/propagate_constraints the solver + // narrows x to roughly [11.0, 16.0]: + // ceil(x) > 12 → x ∈ (11, 16] → conservative [11, 16] + // selectivity ≈ (16−11)/(16−8) = 5/8 = 0.625 → ~62 rows + // + // Without the fix the estimate stays at 100 (no interval analysis). + let schema = Schema::new(vec![Field::new("x", DataType::Float64, false)]); + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(100), + total_byte_size: Precision::Absent, + column_statistics: vec![ColumnStatistics { + min_value: Precision::Inexact(ScalarValue::Float64(Some(8.0))), + max_value: Precision::Inexact(ScalarValue::Float64(Some(16.0))), + ..Default::default() + }], + }, + schema.clone(), + )); + + let x = col("x", &schema)?; + let ceil_udf = datafusion_functions::math::ceil(); + let config = Arc::new(ConfigOptions::new()); + let ceil_x: Arc = + Arc::new(datafusion_physical_expr::ScalarFunctionExpr::try_new( + Arc::clone(&ceil_udf), + vec![x], + &schema, + config, + )?); + let predicate = binary(ceil_x, Operator::Gt, lit(12.0f64), &schema)?; + + let filter = Arc::new(FilterExec::try_new(predicate, input)?); + let statistics = filter.partition_statistics(None)?; + + let num_rows = statistics.num_rows.get_value().copied().unwrap_or(100); + // Interval analysis must narrow the estimate below the full 100-row input. + assert!( + num_rows < 100, + "expected interval analysis to narrow row estimate, got {num_rows}" + ); + // The conservative bound is x ∈ [11, 16] out of [8, 16] → ~62 rows. + // Allow a generous range to be robust to float-cardinality rounding. + assert!( + num_rows >= 50, + "expected at least 50 rows after ceil(x) > 12.0 on [8,16], got {num_rows}" + ); + Ok(()) + } + #[tokio::test] async fn test_filter_statistics_basic_expr() -> Result<()> { // Table: From 900634b8e9eec421eb3483dc547cb637206a8025 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sat, 16 May 2026 15:49:27 +0200 Subject: [PATCH 03/14] fix copilot --- datafusion/functions/src/math/ceil.rs | 10 ++- .../physical-expr/src/intervals/utils.rs | 63 +++++++++++++++---- 2 files changed, 57 insertions(+), 16 deletions(-) diff --git a/datafusion/functions/src/math/ceil.rs b/datafusion/functions/src/math/ceil.rs index b67699a8bd7d9..7927272917b4f 100644 --- a/datafusion/functions/src/math/ceil.rs +++ b/datafusion/functions/src/math/ceil.rs @@ -192,7 +192,11 @@ impl ScalarUDFImpl for CeilFunc { fn evaluate_bounds(&self, inputs: &[&Interval]) -> Result { let [input] = inputs else { - return Interval::make_unbounded(&DataType::Float64); + let data_type = inputs + .first() + .map(|i| i.data_type()) + .unwrap_or(DataType::Float64); + return Interval::make_unbounded(&data_type); }; let data_type = input.data_type(); match (ceil_scalar(input.lower()), ceil_scalar(input.upper())) { @@ -208,7 +212,7 @@ impl ScalarUDFImpl for CeilFunc { inputs: &[&Interval], ) -> Result>> { let [input_interval] = inputs else { - return Ok(Some(vec![])); + return Ok(Some(inputs.iter().map(|i| (*i).clone()).collect())); }; // ceil(x) ∈ [N, M] → x ∈ (N−1, M] — conservative closed: [N−1, M] let lo = match interval.lower() { @@ -234,7 +238,7 @@ impl ScalarUDFImpl for CeilFunc { let constraint = Interval::try_new(lo, hi)?; Ok(input_interval.intersect(constraint)?.map(|r| vec![r])) } - _ => Ok(Some(vec![])), + _ => Ok(Some(vec![(*input_interval).clone()])), } } diff --git a/datafusion/physical-expr/src/intervals/utils.rs b/datafusion/physical-expr/src/intervals/utils.rs index b62b6c260f5c7..3a37fb306f21d 100644 --- a/datafusion/physical-expr/src/intervals/utils.rs +++ b/datafusion/physical-expr/src/intervals/utils.rs @@ -207,8 +207,11 @@ mod tests { use crate::expressions::{Column, Literal}; use crate::scalar_function::ScalarFunctionExpr; use arrow::datatypes::{Field, Schema}; - use datafusion_common::ScalarValue; use datafusion_common::config::ConfigOptions; + use datafusion_common::{Result, ScalarValue}; + use datafusion_expr::{ + ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility, + }; fn f64_schema() -> SchemaRef { Arc::new(Schema::new(vec![Field::new("x", DataType::Float64, false)])) @@ -241,6 +244,48 @@ mod tests { )) } + /// A minimal UDF whose declared return type is Utf8, used to test that + /// check_support rejects functions with unsupported return types without + /// relying on an invalid ceil-returns-Utf8 combination. + #[derive(Debug, PartialEq, Eq, Hash)] + struct Utf8UDF { + signature: Signature, + } + + impl Utf8UDF { + fn new() -> Self { + Self { + signature: Signature::uniform( + 1, + vec![DataType::Float64], + Volatility::Immutable, + ), + } + } + } + + impl ScalarUDFImpl for Utf8UDF { + fn as_any(&self) -> &dyn std::any::Any { + self + } + fn name(&self) -> &str { + "utf8_udf" + } + fn signature(&self) -> &Signature { + &self.signature + } + fn return_type(&self, _: &[DataType]) -> Result { + Ok(DataType::Utf8) + } + fn invoke_with_args(&self, _: ScalarFunctionArgs) -> Result { + Ok(ColumnarValue::Scalar(ScalarValue::Utf8(None))) + } + } + + fn utf8_udf() -> Arc { + Arc::new(datafusion_expr::ScalarUDF::from(Utf8UDF::new())) + } + #[test] fn test_check_support_scalar_fn_supported_return_type() { // ceil(x) returns Float64 — both return type and child are supported @@ -255,13 +300,9 @@ mod tests { #[test] fn test_check_support_scalar_fn_unsupported_return_type() { - // A UDF that returns Utf8 — not in is_datatype_supported + // utf8_udf(x) returns Utf8 — not in is_datatype_supported let schema = f64_schema(); - let expr = scalar_fn_expr( - datafusion_functions::math::ceil(), - vec![col_x()], - DataType::Utf8, - ); + let expr = scalar_fn_expr(utf8_udf(), vec![col_x()], DataType::Utf8); assert!(!check_support(&expr, &schema)); } @@ -294,13 +335,9 @@ mod tests { #[test] fn test_check_support_scalar_fn_in_binary_expr_unsupported_return() { - // f(x) > 5.0 where f returns Utf8 — should be false + // utf8_udf(x) > 5.0 where f returns Utf8 — should be false let schema = f64_schema(); - let fn_expr = scalar_fn_expr( - datafusion_functions::math::ceil(), - vec![col_x()], - DataType::Utf8, - ); + let fn_expr = scalar_fn_expr(utf8_udf(), vec![col_x()], DataType::Utf8); let expr: Arc = Arc::new(BinaryExpr::new(fn_expr, Operator::Gt, lit_f64(5.0))); assert!(!check_support(&expr, &schema)); From 3654e18ef425c1448e7fac34cb4907f5113916c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sat, 16 May 2026 15:57:40 +0200 Subject: [PATCH 04/14] fix copilot --- datafusion/functions/src/math/ceil.rs | 14 ++++++++------ datafusion/physical-expr/src/intervals/utils.rs | 8 +++----- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/datafusion/functions/src/math/ceil.rs b/datafusion/functions/src/math/ceil.rs index 7927272917b4f..f2aa879bd8b11 100644 --- a/datafusion/functions/src/math/ceil.rs +++ b/datafusion/functions/src/math/ceil.rs @@ -192,11 +192,10 @@ impl ScalarUDFImpl for CeilFunc { fn evaluate_bounds(&self, inputs: &[&Interval]) -> Result { let [input] = inputs else { - let data_type = inputs - .first() - .map(|i| i.data_type()) - .unwrap_or(DataType::Float64); - return Interval::make_unbounded(&data_type); + return exec_err!( + "ceil expected 1 argument for bounds evaluation, got {}", + inputs.len() + ); }; let data_type = input.data_type(); match (ceil_scalar(input.lower()), ceil_scalar(input.upper())) { @@ -212,7 +211,10 @@ impl ScalarUDFImpl for CeilFunc { inputs: &[&Interval], ) -> Result>> { let [input_interval] = inputs else { - return Ok(Some(inputs.iter().map(|i| (*i).clone()).collect())); + return exec_err!( + "ceil expected 1 argument for constraint propagation, got {}", + inputs.len() + ); }; // ceil(x) ∈ [N, M] → x ∈ (N−1, M] — conservative closed: [N−1, M] let lo = match interval.lower() { diff --git a/datafusion/physical-expr/src/intervals/utils.rs b/datafusion/physical-expr/src/intervals/utils.rs index 3a37fb306f21d..4027f65b1cdd7 100644 --- a/datafusion/physical-expr/src/intervals/utils.rs +++ b/datafusion/physical-expr/src/intervals/utils.rs @@ -28,8 +28,8 @@ use crate::{ use arrow::array::types::{IntervalDayTime, IntervalMonthDayNano}; use arrow::datatypes::{DataType, SchemaRef}; use datafusion_common::{Result, ScalarValue, internal_err}; -use datafusion_expr::Operator; use datafusion_expr::interval_arithmetic::Interval; +use datafusion_expr::{Operator, Volatility}; /// Indicates whether interval arithmetic is supported for the given expression. /// Currently, we do not support all [`PhysicalExpr`]s for interval calculations. @@ -58,7 +58,8 @@ pub fn check_support(expr: &Arc, schema: &SchemaRef) -> bool { } else if let Some(negative) = expr.downcast_ref::() { check_support(negative.arg(), schema) } else if let Some(scalar_fn) = expr.downcast_ref::() { - is_datatype_supported(scalar_fn.return_type()) + scalar_fn.fun().signature().volatility == Volatility::Immutable + && is_datatype_supported(scalar_fn.return_type()) && scalar_fn .args() .iter() @@ -265,9 +266,6 @@ mod tests { } impl ScalarUDFImpl for Utf8UDF { - fn as_any(&self) -> &dyn std::any::Any { - self - } fn name(&self) -> &str { "utf8_udf" } From ceca4f49326d0ba928a6bd3543334efb6ebcbe46 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sat, 16 May 2026 16:33:24 +0200 Subject: [PATCH 05/14] fix: address review feedback on interval analysis for ScalarFunctionExpr --- datafusion/expr/src/udf.rs | 4 ++-- datafusion/physical-expr/src/scalar_function.rs | 7 ++++++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/datafusion/expr/src/udf.rs b/datafusion/expr/src/udf.rs index 6a3aa31a8609f..8276dfe224288 100644 --- a/datafusion/expr/src/udf.rs +++ b/datafusion/expr/src/udf.rs @@ -929,9 +929,9 @@ pub trait ScalarUDFImpl: Debug + DynEq + DynHash + Send + Sync + Any { fn propagate_constraints( &self, _interval: &Interval, - _inputs: &[&Interval], + inputs: &[&Interval], ) -> Result>> { - Ok(Some(vec![])) + Ok(Some(inputs.iter().map(|i| (*i).clone()).collect())) } /// Calculates the [`SortProperties`] of this function based on its children's properties. diff --git a/datafusion/physical-expr/src/scalar_function.rs b/datafusion/physical-expr/src/scalar_function.rs index 418d005c971ea..b81d4254ede34 100644 --- a/datafusion/physical-expr/src/scalar_function.rs +++ b/datafusion/physical-expr/src/scalar_function.rs @@ -302,7 +302,12 @@ impl PhysicalExpr for ScalarFunctionExpr { } fn evaluate_bounds(&self, children: &[&Interval]) -> Result { - self.fun.evaluate_bounds(children) + let result = self.fun.evaluate_bounds(children)?; + if result.data_type() == DataType::Null { + Interval::make_unbounded(self.return_type()) + } else { + Ok(result) + } } fn propagate_constraints( From 0916e2a096b6560d9ea897b806dd90fcf495b812 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sat, 16 May 2026 17:15:00 +0200 Subject: [PATCH 06/14] fix(ceil): tighten propagate_constraints bounds using integer normalization --- datafusion/functions/src/math/ceil.rs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/datafusion/functions/src/math/ceil.rs b/datafusion/functions/src/math/ceil.rs index f2aa879bd8b11..67d7a75bd0c17 100644 --- a/datafusion/functions/src/math/ceil.rs +++ b/datafusion/functions/src/math/ceil.rs @@ -216,22 +216,23 @@ impl ScalarUDFImpl for CeilFunc { inputs.len() ); }; - // ceil(x) ∈ [N, M] → x ∈ (N−1, M] — conservative closed: [N−1, M] + // ceil(x) ∈ [N, M] → x ∈ (ceil(N)−1, floor(M)] + // Normalize bounds to integers ceil can actually take before mapping back. let lo = match interval.lower() { ScalarValue::Float64(Some(n)) if n.is_finite() => { - Some(ScalarValue::Float64(Some(n - 1.0))) + Some(ScalarValue::Float64(Some(n.ceil() - 1.0))) } ScalarValue::Float32(Some(n)) if n.is_finite() => { - Some(ScalarValue::Float32(Some(n - 1.0))) + Some(ScalarValue::Float32(Some(n.ceil() - 1.0))) } _ => None, }; let hi = match interval.upper() { ScalarValue::Float64(Some(n)) if n.is_finite() => { - Some(ScalarValue::Float64(Some(*n))) + Some(ScalarValue::Float64(Some(n.floor()))) } ScalarValue::Float32(Some(n)) if n.is_finite() => { - Some(ScalarValue::Float32(Some(*n))) + Some(ScalarValue::Float32(Some(n.floor()))) } _ => None, }; From 5412343b007e8c8215295697eabf2d9c53d8e5b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sat, 16 May 2026 21:39:04 +0200 Subject: [PATCH 07/14] tests --- datafusion/expr/src/udf.rs | 3 + datafusion/functions/src/math/ceil.rs | 154 ++++++++++++++++++++++++++ 2 files changed, 157 insertions(+) diff --git a/datafusion/expr/src/udf.rs b/datafusion/expr/src/udf.rs index 8276dfe224288..f9b7b72c341f8 100644 --- a/datafusion/expr/src/udf.rs +++ b/datafusion/expr/src/udf.rs @@ -931,6 +931,9 @@ pub trait ScalarUDFImpl: Debug + DynEq + DynHash + Send + Sync + Any { _interval: &Interval, inputs: &[&Interval], ) -> Result>> { + // Conservative default: return inputs unchanged (no narrowing). + // The returned vec must have the same length as `inputs` to satisfy + // the interval solver contract. Ok(Some(inputs.iter().map(|i| (*i).clone()).collect())) } diff --git a/datafusion/functions/src/math/ceil.rs b/datafusion/functions/src/math/ceil.rs index 67d7a75bd0c17..d431a37383567 100644 --- a/datafusion/functions/src/math/ceil.rs +++ b/datafusion/functions/src/math/ceil.rs @@ -261,3 +261,157 @@ fn ceil_scalar(v: &ScalarValue) -> Option { _ => None, } } + +#[cfg(test)] +mod tests { + use super::*; + + fn ceil() -> CeilFunc { + CeilFunc::new() + } + + fn f64_interval(lo: f64, hi: f64) -> Interval { + Interval::try_new( + ScalarValue::Float64(Some(lo)), + ScalarValue::Float64(Some(hi)), + ) + .unwrap() + } + + fn f32_interval(lo: f32, hi: f32) -> Interval { + Interval::try_new( + ScalarValue::Float32(Some(lo)), + ScalarValue::Float32(Some(hi)), + ) + .unwrap() + } + + fn unbounded_f64() -> Interval { + Interval::make_unbounded(&DataType::Float64).unwrap() + } + + fn unbounded_f32() -> Interval { + Interval::make_unbounded(&DataType::Float32).unwrap() + } + + // --- evaluate_bounds --- + + #[test] + fn test_evaluate_bounds_basic() { + // ceil([1.2, 3.7]) = [2.0, 4.0] + let input = f64_interval(1.2, 3.7); + let result = ceil().evaluate_bounds(&[&input]).unwrap(); + assert_eq!(result, f64_interval(2.0, 4.0)); + } + + #[test] + fn test_evaluate_bounds_already_integer() { + // ceil([2.0, 4.0]) = [2.0, 4.0] + let input = f64_interval(2.0, 4.0); + let result = ceil().evaluate_bounds(&[&input]).unwrap(); + assert_eq!(result, f64_interval(2.0, 4.0)); + } + + #[test] + fn test_evaluate_bounds_f32() { + // ceil([1.1f32, 2.9f32]) = [2.0f32, 3.0f32] + let input = f32_interval(1.1, 2.9); + let result = ceil().evaluate_bounds(&[&input]).unwrap(); + assert_eq!(result, f32_interval(2.0, 3.0)); + } + + #[test] + fn test_evaluate_bounds_unbounded_returns_unbounded() { + let input = unbounded_f64(); + let result = ceil().evaluate_bounds(&[&input]).unwrap(); + assert_eq!(result, unbounded_f64()); + } + + #[test] + fn test_evaluate_bounds_negative() { + // ceil([-3.7, -1.2]) = [-3.0, -1.0] + let input = f64_interval(-3.7, -1.2); + let result = ceil().evaluate_bounds(&[&input]).unwrap(); + assert_eq!(result, f64_interval(-3.0, -1.0)); + } + + // --- propagate_constraints --- + + #[test] + fn test_propagate_constraints_basic() { + // ceil(x) ∈ [13.0, 15.0] → x ∈ (12.0, 15.0] + let output = f64_interval(13.0, 15.0); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(12.0, 15.0)); + } + + #[test] + fn test_propagate_constraints_non_integer_bounds() { + // ceil(x) ∈ [12.3, 14.7] — non-integer bounds are normalized: + // lower: ceil(12.3)-1 = 13-1 = 12.0, upper: floor(14.7) = 14.0 + // → x ∈ (12.0, 14.0] + let output = f64_interval(12.3, 14.7); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(12.0, 14.0)); + } + + #[test] + fn test_propagate_constraints_f32() { + // Same as basic but with Float32 + let output = f32_interval(5.0, 8.0); + let input = unbounded_f32(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f32_interval(4.0, 8.0)); + } + + #[test] + fn test_propagate_constraints_unbounded_output_no_change() { + // No output constraint → input unchanged + let output = unbounded_f64(); + let input = f64_interval(1.0, 10.0); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], input); + } + + #[test] + fn test_propagate_constraints_nan_output_no_change() { + // NaN bounds → conservative: input unchanged + let output = Interval::try_new( + ScalarValue::Float64(Some(f64::NAN)), + ScalarValue::Float64(Some(f64::NAN)), + ) + .unwrap(); + let input = f64_interval(0.0, 100.0); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], input); + } + + #[test] + fn test_propagate_constraints_negative_range() { + // ceil(x) ∈ [-3.0, -1.0] → x ∈ (-4.0, -1.0] + let output = f64_interval(-3.0, -1.0); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(-4.0, -1.0)); + } +} From ade81380cc320e4299da36705da1f9708cc0fab5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sat, 16 May 2026 21:44:18 +0200 Subject: [PATCH 08/14] test --- datafusion/functions/src/math/ceil.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/datafusion/functions/src/math/ceil.rs b/datafusion/functions/src/math/ceil.rs index d431a37383567..7d8aca11289d0 100644 --- a/datafusion/functions/src/math/ceil.rs +++ b/datafusion/functions/src/math/ceil.rs @@ -414,4 +414,17 @@ mod tests { .unwrap(); assert_eq!(result[0], f64_interval(-4.0, -1.0)); } + + #[test] + fn test_propagate_constraints_empty_intersection() { + // x ∈ [5.0, 7.0], constraint ceil(x) ∈ [20.0, 30.0] + // mapped input constraint: [19.0, 30.0] — no overlap with [5.0, 7.0] + // → intersect returns None → Ok(None) (branch pruned) + let output = f64_interval(20.0, 30.0); + let input = f64_interval(5.0, 7.0); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap(); + assert!(result.is_none()); + } } From 7f73e239eca56deea9178618259dece7d8181efe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sat, 16 May 2026 22:04:40 +0200 Subject: [PATCH 09/14] fmt --- datafusion/functions/src/math/ceil.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/datafusion/functions/src/math/ceil.rs b/datafusion/functions/src/math/ceil.rs index 7d8aca11289d0..e4914a89b9423 100644 --- a/datafusion/functions/src/math/ceil.rs +++ b/datafusion/functions/src/math/ceil.rs @@ -422,9 +422,7 @@ mod tests { // → intersect returns None → Ok(None) (branch pruned) let output = f64_interval(20.0, 30.0); let input = f64_interval(5.0, 7.0); - let result = ceil() - .propagate_constraints(&output, &[&input]) - .unwrap(); + let result = ceil().propagate_constraints(&output, &[&input]).unwrap(); assert!(result.is_none()); } } From afc1d556e562e4351b14d03d9e3f9584c70fd4f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sun, 7 Jun 2026 19:56:31 +0200 Subject: [PATCH 10/14] fix copilot --- datafusion/functions/src/math/ceil.rs | 49 +++++++++++++++++++++++--- datafusion/physical-plan/src/filter.rs | 16 ++++++--- 2 files changed, 55 insertions(+), 10 deletions(-) diff --git a/datafusion/functions/src/math/ceil.rs b/datafusion/functions/src/math/ceil.rs index e4914a89b9423..835f9cb928bbb 100644 --- a/datafusion/functions/src/math/ceil.rs +++ b/datafusion/functions/src/math/ceil.rs @@ -199,8 +199,9 @@ impl ScalarUDFImpl for CeilFunc { }; let data_type = input.data_type(); match (ceil_scalar(input.lower()), ceil_scalar(input.upper())) { - (Some(lo), Some(hi)) => Interval::try_new(lo, hi) - .or_else(|_| Interval::make_unbounded(&data_type)), + // Both bounds finite → ceil preserves type and monotonicity, so + // `try_new` always succeeds; let it propagate if that ever breaks. + (Some(lo), Some(hi)) => Interval::try_new(lo, hi), _ => Interval::make_unbounded(&data_type), } } @@ -236,12 +237,18 @@ impl ScalarUDFImpl for CeilFunc { } _ => None, }; + // If either side of the output is finite we can still narrow that side + // of the input; the unknown side falls back to the input's own bound so + // the intersect is a no-op on it. match (lo, hi) { - (Some(lo), Some(hi)) => { - let constraint = Interval::try_new(lo, hi)?; + (None, None) => Ok(Some(vec![(*input_interval).clone()])), + (lo, hi) => { + let constraint = Interval::try_new( + lo.unwrap_or_else(|| input_interval.lower().clone()), + hi.unwrap_or_else(|| input_interval.upper().clone()), + )?; Ok(input_interval.intersect(constraint)?.map(|r| vec![r])) } - _ => Ok(Some(vec![(*input_interval).clone()])), } } @@ -415,6 +422,38 @@ mod tests { assert_eq!(result[0], f64_interval(-4.0, -1.0)); } + #[test] + fn test_propagate_constraints_one_sided_upper() { + // Output (-Inf, 3.5] → x ≤ floor(3.5) = 3.0; input [0.0, 10.0] → [0.0, 3.0] + let output = Interval::try_new( + ScalarValue::Float64(None), + ScalarValue::Float64(Some(3.5)), + ) + .unwrap(); + let input = f64_interval(0.0, 10.0); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(0.0, 3.0)); + } + + #[test] + fn test_propagate_constraints_one_sided_lower() { + // Output [5.0, +Inf) → x > ceil(5.0) - 1 = 4.0; input [0.0, 10.0] → [4.0, 10.0] + let output = Interval::try_new( + ScalarValue::Float64(Some(5.0)), + ScalarValue::Float64(None), + ) + .unwrap(); + let input = f64_interval(0.0, 10.0); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(4.0, 10.0)); + } + #[test] fn test_propagate_constraints_empty_intersection() { // x ∈ [5.0, 7.0], constraint ceil(x) ∈ [20.0, 30.0] diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 5e4417f45b6ae..3666c69d0ff89 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -1360,17 +1360,23 @@ mod tests { let filter = Arc::new(FilterExec::try_new(predicate, input)?); let statistics = filter.partition_statistics(None)?; + let input_num_rows = 100.0_f64; let num_rows = statistics.num_rows.get_value().copied().unwrap_or(100); // Interval analysis must narrow the estimate below the full 100-row input. assert!( - num_rows < 100, + (num_rows as f64) < input_num_rows, "expected interval analysis to narrow row estimate, got {num_rows}" ); - // The conservative bound is x ∈ [11, 16] out of [8, 16] → ~62 rows. - // Allow a generous range to be robust to float-cardinality rounding. + // Conservative bound x ∈ [11, 16] out of [8, 16] → selectivity 5/8 ≈ 62 + // rows. Derive the expected value from the formula and allow a 20% + // tolerance to absorb float-cardinality rounding without masking real + // regressions. + let expected_rows = input_num_rows * (16.0 - 11.0) / (16.0 - 8.0); + let lower_bound = expected_rows * 0.8; assert!( - num_rows >= 50, - "expected at least 50 rows after ceil(x) > 12.0 on [8,16], got {num_rows}" + (num_rows as f64) >= lower_bound, + "expected materially narrowed estimate after ceil(x) > 12.0 on \ + [8,16]; expected at least {lower_bound:.1}, got {num_rows}" ); Ok(()) } From 3cf6a22e3ab82ad1200585797371ccdc8f0d7b48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sun, 7 Jun 2026 20:08:31 +0200 Subject: [PATCH 11/14] refactor: accept Stable UDFs in interval analysis, clarify boundaries --- .../physical-expr/src/intervals/utils.rs | 73 ++++++++++++++++++- .../physical-expr/src/scalar_function.rs | 5 ++ .../physical-plan/src/aggregates/row_hash.rs | 4 + 3 files changed, 81 insertions(+), 1 deletion(-) diff --git a/datafusion/physical-expr/src/intervals/utils.rs b/datafusion/physical-expr/src/intervals/utils.rs index 4027f65b1cdd7..7c7da49b541d9 100644 --- a/datafusion/physical-expr/src/intervals/utils.rs +++ b/datafusion/physical-expr/src/intervals/utils.rs @@ -58,7 +58,11 @@ pub fn check_support(expr: &Arc, schema: &SchemaRef) -> bool { } else if let Some(negative) = expr.downcast_ref::() { check_support(negative.arg(), schema) } else if let Some(scalar_fn) = expr.downcast_ref::() { - scalar_fn.fun().signature().volatility == Volatility::Immutable + // Stable functions (e.g. `now()`, `current_date()`) are deterministic + // within a single query execution, so their bounds are well-defined for + // interval analysis — only Volatile functions (e.g. `random()`) must be + // excluded. + scalar_fn.fun().signature().volatility != Volatility::Volatile && is_datatype_supported(scalar_fn.return_type()) && scalar_fn .args() @@ -340,4 +344,71 @@ mod tests { Arc::new(BinaryExpr::new(fn_expr, Operator::Gt, lit_f64(5.0))); assert!(!check_support(&expr, &schema)); } + + /// A Float64-returning UDF parameterised by volatility, used to verify + /// that `check_support` accepts Immutable/Stable and rejects Volatile. + #[derive(Debug, PartialEq, Eq, Hash)] + struct VolatilityUDF { + name: &'static str, + signature: Signature, + } + + impl VolatilityUDF { + fn new(name: &'static str, volatility: Volatility) -> Self { + Self { + name, + signature: Signature::uniform(1, vec![DataType::Float64], volatility), + } + } + } + + impl ScalarUDFImpl for VolatilityUDF { + fn name(&self) -> &str { + self.name + } + fn signature(&self) -> &Signature { + &self.signature + } + fn return_type(&self, _: &[DataType]) -> Result { + Ok(DataType::Float64) + } + fn invoke_with_args(&self, _: ScalarFunctionArgs) -> Result { + Ok(ColumnarValue::Scalar(ScalarValue::Float64(None))) + } + } + + fn volatility_udf( + name: &'static str, + volatility: Volatility, + ) -> Arc { + Arc::new(datafusion_expr::ScalarUDF::from(VolatilityUDF::new( + name, volatility, + ))) + } + + #[test] + fn test_check_support_scalar_fn_stable_is_supported() { + // Stable functions are deterministic within a query → safe for interval + // analysis; check_support must accept them. + let schema = f64_schema(); + let expr = scalar_fn_expr( + volatility_udf("stable_udf", Volatility::Stable), + vec![col_x()], + DataType::Float64, + ); + assert!(check_support(&expr, &schema)); + } + + #[test] + fn test_check_support_scalar_fn_volatile_is_unsupported() { + // Volatile functions (e.g. random()) can return different values on each + // call → interval analysis is unsound; check_support must reject them. + let schema = f64_schema(); + let expr = scalar_fn_expr( + volatility_udf("volatile_udf", Volatility::Volatile), + vec![col_x()], + DataType::Float64, + ); + assert!(!check_support(&expr, &schema)); + } } diff --git a/datafusion/physical-expr/src/scalar_function.rs b/datafusion/physical-expr/src/scalar_function.rs index b81d4254ede34..d750db27041be 100644 --- a/datafusion/physical-expr/src/scalar_function.rs +++ b/datafusion/physical-expr/src/scalar_function.rs @@ -302,6 +302,11 @@ impl PhysicalExpr for ScalarFunctionExpr { } fn evaluate_bounds(&self, children: &[&Interval]) -> Result { + // The `ScalarUDFImpl::evaluate_bounds` default cannot know the actual + // return type and falls back to an unbounded `Null` interval. Re-type it + // here using the resolved return type so downstream solvers receive a + // well-typed interval. UDFs that override `evaluate_bounds` with a real + // (non-Null) return are passed through untouched. let result = self.fun.evaluate_bounds(children)?; if result.data_type() == DataType::Null { Interval::make_unbounded(self.return_type()) diff --git a/datafusion/physical-plan/src/aggregates/row_hash.rs b/datafusion/physical-plan/src/aggregates/row_hash.rs index c3f73976c721a..85dce2660e1c1 100644 --- a/datafusion/physical-plan/src/aggregates/row_hash.rs +++ b/datafusion/physical-plan/src/aggregates/row_hash.rs @@ -194,6 +194,10 @@ impl SkipAggregationProbe { self.input_rows += input_rows; self.num_groups = num_groups; if self.input_rows >= self.probe_rows_threshold { + // Strict `>` (not `>=`): the boundary `ratio == threshold` must NOT + // trigger skip — without this, threshold=1.0 (100% cardinality, the + // documented way to disable the feature) would still cause skipping + // when every row is its own group. self.should_skip = self.num_groups as f64 / self.input_rows as f64 > self.probe_ratio_threshold; // Set is_locked to true only if we have decided to skip, otherwise we can try to skip From 40cddef3436ff87a7e35a6a431cb58d0a718a0e3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Wed, 10 Jun 2026 20:48:21 +0200 Subject: [PATCH 12/14] prune empty preimages in ceil propagate_constraints --- datafusion/functions/src/math/ceil.rs | 299 ++++++++++++++++++++++++++ 1 file changed, 299 insertions(+) diff --git a/datafusion/functions/src/math/ceil.rs b/datafusion/functions/src/math/ceil.rs index 835f9cb928bbb..89ab8a33e6dc9 100644 --- a/datafusion/functions/src/math/ceil.rs +++ b/datafusion/functions/src/math/ceil.rs @@ -237,6 +237,17 @@ impl ScalarUDFImpl for CeilFunc { } _ => None, }; + // When BOTH bounds of the output are finite, the output interval + // [N, M] admits an integer (and therefore a preimage) only when + // `ceil(N) ≤ floor(M)`. With our transformed values that's + // `lo + 1 ≤ hi`, i.e. `lo < hi`. If `lo ≥ hi`, the output contains + // no integer (e.g. `ceil(x) ∈ [12.3, 12.7]` or `[13.1, 13.1]`), the + // preimage is empty, and we can prune the branch. + if let (Some(l), Some(h)) = (&lo, &hi) + && l >= h + { + return Ok(None); + } // If either side of the output is finite we can still narrow that side // of the input; the unknown side falls back to the input's own bound so // the intersect is a no-op on it. @@ -257,6 +268,12 @@ impl ScalarUDFImpl for CeilFunc { } } +/// IEEE-754 signed-zero is intentionally preserved: `(-0.001).ceil() == -0.0` +/// (not `+0.0`), and `ScalarValue` compares bit-for-bit, so `Float64(-0.0)` is +/// structurally distinct from `Float64(+0.0)` even though numerically equal. +/// Do not "normalise" `-0.0 → +0.0` here without considering downstream +/// hashing, structural equality, and Arrow consistency — see the +/// `test_*_zero` and `test_*_singleton_negative_zero` regression tests. fn ceil_scalar(v: &ScalarValue) -> Option { match v { ScalarValue::Float64(Some(f)) if f.is_finite() => { @@ -464,4 +481,286 @@ mod tests { let result = ceil().propagate_constraints(&output, &[&input]).unwrap(); assert!(result.is_none()); } + + // --- Output contains no integer → preimage is empty, branch prunes --- + + /// `ceil(x) ∈ [12.3, 12.7]` — both bounds non-integer in the same gap + /// (12, 13). No integer satisfies `ceil(x) ∈ [12.3, 12.7]`, so the + /// preimage is empty and the branch must be pruned (`Ok(None)`). + #[test] + fn test_propagate_constraints_no_integer_in_output_positive_width() { + let output = f64_interval(12.3, 12.7); + let input = unbounded_f64(); + let result = ceil().propagate_constraints(&output, &[&input]).unwrap(); + assert!(result.is_none(), "expected branch pruned, got {result:?}"); + } + + /// `ceil(x) ∈ [13.1, 13.1]` — degenerate (singleton) non-integer interval. + /// Same logic: `ceil` only produces integers, so no x maps into a + /// non-integer singleton. Branch must be pruned. + #[test] + fn test_propagate_constraints_no_integer_in_output_degenerate() { + let output = f64_interval(13.1, 13.1); + let input = unbounded_f64(); + let result = ceil().propagate_constraints(&output, &[&input]).unwrap(); + assert!(result.is_none(), "expected branch pruned, got {result:?}"); + } + + /// `ceil(x) ∈ [13.1, 13.9]` — wider non-integer interval that still + /// contains no integer. Branch must be pruned. + #[test] + fn test_propagate_constraints_no_integer_in_output_wider_gap() { + let output = f64_interval(13.1, 13.9); + let input = unbounded_f64(); + let result = ceil().propagate_constraints(&output, &[&input]).unwrap(); + assert!(result.is_none(), "expected branch pruned, got {result:?}"); + } + + /// Float32 variant of the impossibility detection — the same logic must + /// apply regardless of float width. + #[test] + fn test_propagate_constraints_no_integer_in_output_f32() { + let output = f32_interval(7.2, 7.8); + let input = unbounded_f32(); + let result = ceil().propagate_constraints(&output, &[&input]).unwrap(); + assert!(result.is_none(), "expected branch pruned, got {result:?}"); + } + + // --- Cases that must NOT trigger the impossibility check --- + + /// Integer singleton `ceil(x) ∈ [12.0, 12.0]` is feasible: `ceil(12) = 12`. + /// The check `lo >= hi` becomes `11 >= 12` (false), so we DO NOT prune. + #[test] + fn test_propagate_constraints_integer_singleton_is_feasible() { + let output = f64_interval(12.0, 12.0); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(11.0, 12.0)); + } + + /// Boundary case `ceil(x) ∈ [12.0, 12.7]` — N integer, M non-integer in + /// gap (12, 13). Feasible (x=12 → ceil=12 ∈ [12, 12.7]). Must NOT prune. + /// Without the guard the check could over-fire if integer N is mishandled. + #[test] + fn test_propagate_constraints_integer_lower_non_integer_upper_feasible() { + let output = f64_interval(12.0, 12.7); + let input = f64_interval(0.0, 100.0); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + // lo = ceil(12)-1 = 11, hi = floor(12.7) = 12 → constraint [11, 12] + // intersect with [0, 100] → [11, 12] + assert_eq!(result[0], f64_interval(11.0, 12.0)); + } + + /// Boundary case `ceil(x) ∈ [12.3, 13.0]` — N non-integer, M integer. + /// Feasible (x=13 → ceil=13 ∈ [12.3, 13]). Must NOT prune. + #[test] + fn test_propagate_constraints_non_integer_lower_integer_upper_feasible() { + let output = f64_interval(12.3, 13.0); + let input = f64_interval(0.0, 100.0); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + // lo = ceil(12.3)-1 = 12, hi = floor(13) = 13 → constraint [12, 13] + // intersect with [0, 100] → [12, 13] + assert_eq!(result[0], f64_interval(12.0, 13.0)); + } + + /// One-sided output `(-∞, 12.7]` — even though `floor(12.7) = 12` would + /// look "narrow", the lower side is unbounded so the impossibility check + /// must NOT fire (the guard is `(Some, Some)` only). + #[test] + fn test_propagate_constraints_one_sided_does_not_prune() { + let output = Interval::try_new( + ScalarValue::Float64(None), + ScalarValue::Float64(Some(12.7)), + ) + .unwrap(); + let input = f64_interval(0.0, 100.0); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + // hi = floor(12.7) = 12; lower bound falls back to input lower (0.0) + assert_eq!(result[0], f64_interval(0.0, 12.0)); + } + + // --- evaluate_bounds: intervals straddling integer boundaries (off-by-one + // prone). ceil is monotonic, so ceil([a, b]) = [ceil(a), ceil(b)]. --- + + /// `[1.999, 2.001]` — straddles integer 2. ceil(1.999)=2, ceil(2.001)=3. + #[test] + fn test_evaluate_bounds_straddles_integer() { + let input = f64_interval(1.999, 2.001); + let result = ceil().evaluate_bounds(&[&input]).unwrap(); + assert_eq!(result, f64_interval(2.0, 3.0)); + } + + /// `[-2.001, -1.999]` — negative variant of the same boundary case. + /// ceil(-2.001) = -2, ceil(-1.999) = -1. + #[test] + fn test_evaluate_bounds_straddles_negative_integer() { + let input = f64_interval(-2.001, -1.999); + let result = ceil().evaluate_bounds(&[&input]).unwrap(); + assert_eq!(result, f64_interval(-2.0, -1.0)); + } + + /// `[-0.001, 0.001]` — straddles zero, the sign-flip boundary. + /// Note: `f64::ceil(-0.001)` returns IEEE-754 `-0.0` (sign-preserving), + /// not `+0.0`. Numerically equivalent but bit-level distinct, so the + /// expected lower bound is `-0.0`. + #[test] + fn test_evaluate_bounds_straddles_zero() { + let input = f64_interval(-0.001, 0.001); + let result = ceil().evaluate_bounds(&[&input]).unwrap(); + let expected = Interval::try_new( + ScalarValue::Float64(Some(-0.0)), + ScalarValue::Float64(Some(1.0)), + ) + .unwrap(); + assert_eq!(result, expected); + } + + // --- propagate_constraints: integer singletons at sign-sensitive points --- + + /// `ceil(x) ∈ [0, 0]` — zero is the sign-flip boundary. Real preimage is + /// `(-1, 0]`, conservatively `[-1, 0]`. + #[test] + fn test_propagate_constraints_integer_singleton_zero() { + let output = f64_interval(0.0, 0.0); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(-1.0, 0.0)); + } + + /// `ceil(x) ∈ [-3, -3]` — negative integer singleton. Real preimage is + /// `(-4, -3]`, conservatively `[-4, -3]`. + #[test] + fn test_propagate_constraints_integer_singleton_negative() { + let output = f64_interval(-3.0, -3.0); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(-4.0, -3.0)); + } + + // --- propagate_constraints: single-integer output where the integer is on + // the LOWER bound (the mirror of `non_integer_lower_integer_upper`) --- + + /// `ceil(x) ∈ [13.0, 13.7]` — exactly one integer (13) lies in the output. + /// lo = ceil(13)-1 = 12, hi = floor(13.7) = 13 → constraint [12, 13]. + #[test] + fn test_propagate_constraints_integer_lower_with_room_above() { + let output = f64_interval(13.0, 13.7); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(12.0, 13.0)); + } + + // --- propagate_constraints: multiple integers and broad negative ranges --- + + /// `ceil(x) ∈ [12.3, 15.7]` — three integers (13, 14, 15) in the output. + /// lo = ceil(12.3)-1 = 12, hi = floor(15.7) = 15 → [12, 15]. + #[test] + fn test_propagate_constraints_multiple_integers_in_output() { + let output = f64_interval(12.3, 15.7); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(12.0, 15.0)); + } + + /// `ceil(x) ∈ [-5, -3]` — broader negative range. lo = ceil(-5)-1 = -6, + /// hi = floor(-3) = -3 → [-6, -3]. Mirror of `[3, 5] → [2, 5]`. + #[test] + fn test_propagate_constraints_negative_multi_integer_range() { + let output = f64_interval(-5.0, -3.0); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + assert_eq!(result[0], f64_interval(-6.0, -3.0)); + } + + // --- propagate_constraints: pruning short-circuits intersect with a + // bounded input (no need to compute the intersection) --- + + /// `ceil(x) ∈ [12.3, 12.7]` with a bounded input `[5, 7]`: the new + /// impossibility check fires BEFORE the intersect step, so the result is + /// `Ok(None)` regardless of the input. Distinct from + /// `test_propagate_constraints_empty_intersection`, which exercises the + /// intersect-returns-None path. + #[test] + fn test_propagate_constraints_pruning_short_circuits_intersect() { + let output = f64_interval(12.3, 12.7); + let input = f64_interval(5.0, 7.0); + let result = ceil().propagate_constraints(&output, &[&input]).unwrap(); + assert!( + result.is_none(), + "expected pruning to short-circuit intersect, got {result:?}" + ); + } + + // --- Signed-zero edge cases: ScalarValue compares bit-for-bit, so the + // output preserves the IEEE-754 sign of zero. These tests document + // that behaviour so a future "normalise -0.0 → +0.0" refactor breaks + // them visibly. --- + + /// `ceil(x) ∈ [0.0, 0.0]` (positive zero singleton). Feasible: ceil(0)=0. + /// lo = 0.0.ceil() - 1.0 = -1.0, hi = 0.0.floor() = +0.0 → `[-1.0, +0.0]`. + #[test] + fn test_propagate_constraints_singleton_positive_zero() { + let output = f64_interval(0.0, 0.0); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + let expected = Interval::try_new( + ScalarValue::Float64(Some(-1.0)), + ScalarValue::Float64(Some(0.0)), + ) + .unwrap(); + assert_eq!(result[0], expected); + } + + /// `ceil(x) ∈ [-0.0, -0.0]` (negative zero singleton). Same set + /// numerically as `[+0.0, +0.0]`, but bit-distinct. ceil/floor preserve + /// the negative sign, so the propagated upper bound is `-0.0`, not `+0.0`. + #[test] + fn test_propagate_constraints_singleton_negative_zero() { + let output = Interval::try_new( + ScalarValue::Float64(Some(-0.0)), + ScalarValue::Float64(Some(-0.0)), + ) + .unwrap(); + let input = unbounded_f64(); + let result = ceil() + .propagate_constraints(&output, &[&input]) + .unwrap() + .unwrap(); + let expected = Interval::try_new( + ScalarValue::Float64(Some(-1.0)), + ScalarValue::Float64(Some(-0.0)), + ) + .unwrap(); + assert_eq!(result[0], expected); + } } From faad9fb9639e8716025a6f1b504334e4d04521c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sun, 28 Jun 2026 00:17:13 +0200 Subject: [PATCH 13/14] refactor: keep interval-analysis change focused on scalar functions - Revert the unrelated SkipAggregationProbe `>` comment in skip_partial.rs back to upstream (partial-aggregation skip is out of scope for this change). - Document ScalarFunctionExpr support in the `check_support` doc comment. --- datafusion/physical-expr/src/intervals/utils.rs | 3 ++- datafusion/physical-plan/src/aggregates/skip_partial.rs | 4 ---- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/datafusion/physical-expr/src/intervals/utils.rs b/datafusion/physical-expr/src/intervals/utils.rs index 7c7da49b541d9..ec0295e48bc7d 100644 --- a/datafusion/physical-expr/src/intervals/utils.rs +++ b/datafusion/physical-expr/src/intervals/utils.rs @@ -35,7 +35,8 @@ use datafusion_expr::{Operator, Volatility}; /// Currently, we do not support all [`PhysicalExpr`]s for interval calculations. /// We do not support every type of [`Operator`]s either. Over time, this check /// will relax as more types of `PhysicalExpr`s and `Operator`s are supported. -/// Currently, [`CastExpr`], [`NegativeExpr`], [`BinaryExpr`], [`Column`] and [`Literal`] are supported. +/// Currently, [`CastExpr`], [`NegativeExpr`], [`BinaryExpr`], [`Column`], [`Literal`] and +/// non-volatile [`ScalarFunctionExpr`]s are supported. pub fn check_support(expr: &Arc, schema: &SchemaRef) -> bool { if let Some(binary_expr) = expr.downcast_ref::() { is_operator_supported(binary_expr.op()) diff --git a/datafusion/physical-plan/src/aggregates/skip_partial.rs b/datafusion/physical-plan/src/aggregates/skip_partial.rs index 3b7b838865b89..903235f950f58 100644 --- a/datafusion/physical-plan/src/aggregates/skip_partial.rs +++ b/datafusion/physical-plan/src/aggregates/skip_partial.rs @@ -96,10 +96,6 @@ impl SkipAggregationProbe { self.input_rows += input_rows; self.num_groups = num_groups; if self.input_rows >= self.probe_rows_threshold { - // Strict `>` (not `>=`): the boundary `ratio == threshold` must NOT - // trigger skip — without this, threshold=1.0 (100% cardinality, the - // documented way to disable the feature) would still cause skipping - // when every row is its own group. self.should_skip = self.num_groups as f64 / self.input_rows as f64 > self.probe_ratio_threshold; // Set is_locked to true only if we have decided to skip, otherwise we can try to skip From 7db46b896bb1b4071858f37e147237c381f9454f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?David=20L=C3=B3pez?= Date: Sun, 28 Jun 2026 17:58:18 +0200 Subject: [PATCH 14/14] trigger CI re-run