ceil and test

Author: davidlghellin

datafusion/functions/src/math/ceil.rs

@@ -191,11 +191,66 @@ impl ScalarUDFImpl for CeilFunc {
     }
 
     fn evaluate_bounds(&self, inputs: &[&Interval]) -> Result<Interval> {
-        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<Option<Vec<Interval>>> {
+        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<ScalarValue> {
+    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,
+    }
+}

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<dyn PhysicalExpr> =
+            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: