From c9f25b270d66299561fc1f8b20287fa49d2a83c7 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Fri, 27 Feb 2026 12:13:39 +0000 Subject: [PATCH 01/25] Add ceil --- datafusion/spark/src/function/math/ceil.rs | 204 +++++++++++++++++++++ datafusion/spark/src/function/math/mod.rs | 4 + 2 files changed, 208 insertions(+) create mode 100644 datafusion/spark/src/function/math/ceil.rs diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs new file mode 100644 index 0000000000000..5826d39ae30c7 --- /dev/null +++ b/datafusion/spark/src/function/math/ceil.rs @@ -0,0 +1,204 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::any::Any; +use std::sync::Arc; + +use arrow::array::{AsArray, Decimal128Array}; +use arrow::compute::cast; +use arrow::datatypes::{DataType, Decimal128Type, Float32Type, Float64Type, Int64Type}; +use datafusion_common::utils::take_function_args; +use datafusion_common::{exec_err, Result}; +use datafusion_expr::{ + ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility, +}; + +/// Spark-compatible `ceil` expression +/// +/// +/// Differences with DataFusion ceil: +/// - Spark's ceil returns Int64 for float/integer types +/// - Spark's ceil adjusts precision for Decimal128 types +#[derive(Debug, PartialEq, Eq, Hash)] +pub struct SparkCeil { + signature: Signature, + aliases: Vec, +} + +impl Default for SparkCeil { + fn default() -> Self { + Self::new() + } +} + +impl SparkCeil { + pub fn new() -> Self { + Self { + signature: Signature::numeric(1, Volatility::Immutable), + aliases: vec!["ceiling".to_string()], + } + } +} + +impl ScalarUDFImpl for SparkCeil { + fn as_any(&self) -> &dyn Any { + self + } + + fn name(&self) -> &str { + "ceil" + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, arg_types: &[DataType]) -> Result { + match &arg_types[0] { + DataType::Decimal128(p, s) if *s > 0 => { + let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; + Ok(DataType::Decimal128(new_p, 0)) + } + DataType::Decimal128(p, s) => Ok(DataType::Decimal128(*p, *s)), + _ => Ok(DataType::Int64), + } + } + + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { + let return_type = args.return_type().clone(); + spark_ceil(&args.args, &return_type) + } + + fn aliases(&self) -> &[String] { + &self.aliases + } +} + +fn spark_ceil(args: &[ColumnarValue], return_type: &DataType) -> Result { + let input = match take_function_args("ceil", args)? { + [ColumnarValue::Scalar(value)] => value.to_array()?, + [ColumnarValue::Array(arr)] => Arc::clone(arr), + }; + + let result = match input.data_type() { + DataType::Float32 => Arc::new( + input + .as_primitive::() + .unary::<_, Int64Type>(|x| x.ceil() as i64), + ) as _, + DataType::Float64 => Arc::new( + input + .as_primitive::() + .unary::<_, Int64Type>(|x| x.ceil() as i64), + ) as _, + dt if dt.is_integer() => cast(&input, &DataType::Int64)?, + DataType::Decimal128(_, s) if *s > 0 => { + let div = 10_i128.pow(*s as u32); + let result: Decimal128Array = + input.as_primitive::().unary(|x| { + let d = x / div; + let r = x % div; + if r > 0 { d + 1 } else { d } + }); + Arc::new(result.with_data_type(return_type.clone())) + } + DataType::Decimal128(_, _) => input, + other => return exec_err!("Unsupported data type {other:?} for function ceil"), + }; + + Ok(ColumnarValue::Array(result)) +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::{Decimal128Array, Float32Array, Float64Array, Int64Array}; + use datafusion_common::ScalarValue; + + #[test] + fn test_ceil_float64() { + let input = Float64Array::from(vec![Some(1.1), Some(-1.1), Some(0.0), None]); + let args = vec![ColumnarValue::Array(Arc::new(input))]; + let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = match result { + ColumnarValue::Array(arr) => arr, + _ => panic!("Expected array"), + }; + let result = result.as_primitive::(); + assert_eq!( + result, + &Int64Array::from(vec![Some(2), Some(-1), Some(0), None]) + ); + } + + #[test] + fn test_ceil_float32() { + let input = Float32Array::from(vec![Some(1.5f32), Some(-1.5f32)]); + let args = vec![ColumnarValue::Array(Arc::new(input))]; + let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = match result { + ColumnarValue::Array(arr) => arr, + _ => panic!("Expected array"), + }; + let result = result.as_primitive::(); + assert_eq!(result, &Int64Array::from(vec![Some(2), Some(-1)])); + } + + #[test] + fn test_ceil_int64() { + let input = Int64Array::from(vec![Some(1), Some(-1), None]); + let args = vec![ColumnarValue::Array(Arc::new(input))]; + let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = match result { + ColumnarValue::Array(arr) => arr, + _ => panic!("Expected array"), + }; + let result = result.as_primitive::(); + assert_eq!(result, &Int64Array::from(vec![Some(1), Some(-1), None])); + } + + #[test] + fn test_ceil_decimal128() { + // Decimal128(10, 2): 150 = 1.50, -150 = -1.50, 100 = 1.00 + let return_type = DataType::Decimal128(9, 0); + let input = Decimal128Array::from(vec![Some(150), Some(-150), Some(100), None]) + .with_data_type(DataType::Decimal128(10, 2)); + let args = vec![ColumnarValue::Array(Arc::new(input))]; + let result = spark_ceil(&args, &return_type).unwrap(); + let result = match result { + ColumnarValue::Array(arr) => arr, + _ => panic!("Expected array"), + }; + let result = result.as_primitive::(); + let expected = Decimal128Array::from(vec![Some(2), Some(-1), Some(1), None]) + .with_data_type(return_type); + assert_eq!(result, &expected); + } + + #[test] + fn test_ceil_scalar() { + let input = ScalarValue::Float64(Some(1.1)); + let args = vec![ColumnarValue::Scalar(input)]; + let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = match result { + ColumnarValue::Array(arr) => arr, + _ => panic!("Expected array"), + }; + let result = result.as_primitive::(); + assert_eq!(result, &Int64Array::from(vec![Some(2)])); + } +} diff --git a/datafusion/spark/src/function/math/mod.rs b/datafusion/spark/src/function/math/mod.rs index 7f7d04e06b0be..07c06dcccb68d 100644 --- a/datafusion/spark/src/function/math/mod.rs +++ b/datafusion/spark/src/function/math/mod.rs @@ -17,6 +17,7 @@ pub mod abs; pub mod bin; +pub mod ceil; pub mod expm1; pub mod factorial; pub mod hex; @@ -32,6 +33,7 @@ use datafusion_functions::make_udf_function; use std::sync::Arc; make_udf_function!(abs::SparkAbs, abs); +make_udf_function!(ceil::SparkCeil, ceil); make_udf_function!(expm1::SparkExpm1, expm1); make_udf_function!(factorial::SparkFactorial, factorial); make_udf_function!(hex::SparkHex, hex); @@ -49,6 +51,7 @@ pub mod expr_fn { use datafusion_functions::export_functions; export_functions!((abs, "Returns abs(expr)", arg1)); + export_functions!((ceil, "Returns the smallest integer not less than expr.", arg1)); export_functions!((expm1, "Returns exp(expr) - 1 as a Float64.", arg1)); export_functions!(( factorial, @@ -82,6 +85,7 @@ pub mod expr_fn { pub fn functions() -> Vec> { vec![ abs(), + ceil(), expm1(), factorial(), hex(), From 35adebd02a49f827dcc11dfb1df9275c42b3390a Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Fri, 27 Feb 2026 13:26:42 +0000 Subject: [PATCH 02/25] add comet tests --- datafusion/spark/src/function/math/ceil.rs | 77 +++++++++++++++++++--- 1 file changed, 69 insertions(+), 8 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 5826d39ae30c7..3c025e8171479 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -22,7 +22,7 @@ use arrow::array::{AsArray, Decimal128Array}; use arrow::compute::cast; use arrow::datatypes::{DataType, Decimal128Type, Float32Type, Float64Type, Int64Type}; use datafusion_common::utils::take_function_args; -use datafusion_common::{exec_err, Result}; +use datafusion_common::{Result, exec_err}; use datafusion_expr::{ ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility, }; @@ -131,7 +131,15 @@ mod tests { #[test] fn test_ceil_float64() { - let input = Float64Array::from(vec![Some(1.1), Some(-1.1), Some(0.0), None]); + let input = Float64Array::from(vec![ + Some(125.2345), + Some(15.0001), + Some(0.1), + Some(-0.9), + Some(-1.1), + Some(123.0), + None, + ]); let args = vec![ColumnarValue::Array(Arc::new(input))]; let result = spark_ceil(&args, &DataType::Int64).unwrap(); let result = match result { @@ -141,13 +149,29 @@ mod tests { let result = result.as_primitive::(); assert_eq!( result, - &Int64Array::from(vec![Some(2), Some(-1), Some(0), None]) + &Int64Array::from(vec![ + Some(126), + Some(16), + Some(1), + Some(0), + Some(-1), + Some(123), + None, + ]) ); } #[test] fn test_ceil_float32() { - let input = Float32Array::from(vec![Some(1.5f32), Some(-1.5f32)]); + let input = Float32Array::from(vec![ + Some(125.2345f32), + Some(15.0001f32), + Some(0.1f32), + Some(-0.9f32), + Some(-1.1f32), + Some(123.0f32), + None, + ]); let args = vec![ColumnarValue::Array(Arc::new(input))]; let result = spark_ceil(&args, &DataType::Int64).unwrap(); let result = match result { @@ -155,7 +179,18 @@ mod tests { _ => panic!("Expected array"), }; let result = result.as_primitive::(); - assert_eq!(result, &Int64Array::from(vec![Some(2), Some(-1)])); + assert_eq!( + result, + &Int64Array::from(vec![ + Some(126), + Some(16), + Some(1), + Some(0), + Some(-1), + Some(123), + None, + ]) + ); } #[test] @@ -190,8 +225,34 @@ mod tests { } #[test] - fn test_ceil_scalar() { - let input = ScalarValue::Float64(Some(1.1)); + fn test_ceil_float64_scalar() { + let input = ScalarValue::Float64(Some(-1.1)); + let args = vec![ColumnarValue::Scalar(input)]; + let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = match result { + ColumnarValue::Array(arr) => arr, + _ => panic!("Expected array"), + }; + let result = result.as_primitive::(); + assert_eq!(result, &Int64Array::from(vec![Some(-1)])); + } + + #[test] + fn test_ceil_float32_scalar() { + let input = ScalarValue::Float32(Some(125.2345f32)); + let args = vec![ColumnarValue::Scalar(input)]; + let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = match result { + ColumnarValue::Array(arr) => arr, + _ => panic!("Expected array"), + }; + let result = result.as_primitive::(); + assert_eq!(result, &Int64Array::from(vec![Some(126)])); + } + + #[test] + fn test_ceil_int64_scalar() { + let input = ScalarValue::Int64(Some(48)); let args = vec![ColumnarValue::Scalar(input)]; let result = spark_ceil(&args, &DataType::Int64).unwrap(); let result = match result { @@ -199,6 +260,6 @@ mod tests { _ => panic!("Expected array"), }; let result = result.as_primitive::(); - assert_eq!(result, &Int64Array::from(vec![Some(2)])); + assert_eq!(result, &Int64Array::from(vec![Some(48)])); } } From fb82ec0c7e4a0a3f7da86ca0c22a6711190acfc9 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Fri, 27 Feb 2026 13:32:39 +0000 Subject: [PATCH 03/25] fmt --- datafusion/spark/src/function/math/mod.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/datafusion/spark/src/function/math/mod.rs b/datafusion/spark/src/function/math/mod.rs index 07c06dcccb68d..c14d6cdf50606 100644 --- a/datafusion/spark/src/function/math/mod.rs +++ b/datafusion/spark/src/function/math/mod.rs @@ -51,7 +51,11 @@ pub mod expr_fn { use datafusion_functions::export_functions; export_functions!((abs, "Returns abs(expr)", arg1)); - export_functions!((ceil, "Returns the smallest integer not less than expr.", arg1)); + export_functions!(( + ceil, + "Returns the smallest integer not less than expr.", + arg1 + )); export_functions!((expm1, "Returns exp(expr) - 1 as a Float64.", arg1)); export_functions!(( factorial, From f0d428cd32d928cd9a463572a1d0af7e1b605dbb Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Sat, 28 Feb 2026 09:25:29 +0000 Subject: [PATCH 04/25] Add slt tests --- .../test_files/spark/math/ceil.slt | 99 +++++++++++++++++++ .../test_files/spark/math/ceiling.slt | 16 +++ 2 files changed, 115 insertions(+) diff --git a/datafusion/sqllogictest/test_files/spark/math/ceil.slt b/datafusion/sqllogictest/test_files/spark/math/ceil.slt index c87a29b61fd49..b7cf476fad7c3 100644 --- a/datafusion/sqllogictest/test_files/spark/math/ceil.slt +++ b/datafusion/sqllogictest/test_files/spark/math/ceil.slt @@ -40,3 +40,102 @@ ## PySpark 3.5.5 Result: {'CEIL(5)': 5, 'typeof(CEIL(5))': 'bigint', 'typeof(5)': 'int'} #query #SELECT ceil(5::int); + +# Scalar input: float64 returns bigint +query IIIIIII +SELECT ceil(125.2345::DOUBLE), ceil(15.0001::DOUBLE), ceil(0.1::DOUBLE), ceil(-0.9::DOUBLE), ceil(-1.1::DOUBLE), ceil(123.0::DOUBLE), ceil(NULL::DOUBLE); +---- +126 16 1 0 -1 123 NULL + +# Scalar input: float32 returns bigint +query IIIIIII +SELECT ceil(125.2345::FLOAT), ceil(15.0001::FLOAT), ceil(0.1::FLOAT), ceil(-0.9::FLOAT), ceil(-1.1::FLOAT), ceil(123.0::FLOAT), ceil(NULL::FLOAT); +---- +126 16 1 0 -1 123 NULL + +# Scalar input: integer types all return bigint +query III +SELECT ceil(5::TINYINT), ceil(-3::TINYINT), ceil(NULL::TINYINT); +---- +5 -3 NULL + +query III +SELECT ceil(5::SMALLINT), ceil(-3::SMALLINT), ceil(NULL::SMALLINT); +---- +5 -3 NULL + +query III +SELECT ceil(5::INT), ceil(-3::INT), ceil(NULL::INT); +---- +5 -3 NULL + +query III +SELECT ceil(5::BIGINT), ceil(-3::BIGINT), ceil(NULL::BIGINT); +---- +5 -3 NULL + +# Scalar input: decimal128 with scale > 0 returns decimal with scale 0 +# ceil(1.50) = 2, ceil(-1.50) = -1, ceil(1.00) = 1 +query RRR +SELECT ceil(1.50::DECIMAL(10, 2)), ceil(-1.50::DECIMAL(10, 2)), ceil(1.00::DECIMAL(10, 2)); +---- +2 -1 1 + +# ceil(-0.1) = 0 (smallest positive decimal rounds up to 0 for negatives) +query RR +SELECT ceil(-0.1::DECIMAL(3, 1)), ceil(NULL::DECIMAL(10, 2)); +---- +0 NULL + +# ceil(3.1411) = 4 +query R +SELECT ceil(3.1411::DECIMAL(5, 4)); +---- +4 + +# Scalar input: decimal128 with scale = 0 passes through unchanged +query RRR +SELECT ceil(5::DECIMAL(10, 0)), ceil(-3::DECIMAL(10, 0)), ceil(NULL::DECIMAL(10, 0)); +---- +5 -3 NULL + +# Array input: float64 +query I +SELECT ceil(a) FROM (VALUES (125.2345::DOUBLE), (15.0001::DOUBLE), (0.1::DOUBLE), (-0.9::DOUBLE), (-1.1::DOUBLE), (123.0::DOUBLE), (NULL::DOUBLE)) AS t(a); +---- +126 +16 +1 +0 +-1 +123 +NULL + +# Array input: float32 +query I +SELECT ceil(a) FROM (VALUES (125.2345::FLOAT), (15.0001::FLOAT), (0.1::FLOAT), (-0.9::FLOAT), (-1.1::FLOAT), (123.0::FLOAT), (NULL::FLOAT)) AS t(a); +---- +126 +16 +1 +0 +-1 +123 +NULL + +# Array input: integers +query I +SELECT ceil(a) FROM (VALUES (5::INT), (-3::INT), (NULL::INT)) AS t(a); +---- +5 +-3 +NULL + +# Array input: decimal128 with scale > 0 +query R +SELECT ceil(a) FROM (VALUES (1.50::DECIMAL(10, 2)), (-1.50::DECIMAL(10, 2)), (1.00::DECIMAL(10, 2)), (NULL::DECIMAL(10, 2))) AS t(a); +---- +2 +-1 +1 +NULL diff --git a/datafusion/sqllogictest/test_files/spark/math/ceiling.slt b/datafusion/sqllogictest/test_files/spark/math/ceiling.slt index 2b761faef47df..b68a4be3092de 100644 --- a/datafusion/sqllogictest/test_files/spark/math/ceiling.slt +++ b/datafusion/sqllogictest/test_files/spark/math/ceiling.slt @@ -40,3 +40,19 @@ ## PySpark 3.5.5 Result: {'ceiling(5)': 5, 'typeof(ceiling(5))': 'bigint', 'typeof(5)': 'int'} #query #SELECT ceiling(5::int); + +# ceiling is an alias for ceil +query I +SELECT ceiling(1.5::DOUBLE); +---- +2 + +query I +SELECT ceiling(5::INT); +---- +5 + +query R +SELECT ceiling(1.50::DECIMAL(10, 2)); +---- +2 From 9aad4795a56a7449b0ff8ecd9b7496394ca71c8b Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Sat, 28 Feb 2026 09:31:08 +0000 Subject: [PATCH 05/25] Uncomment existing slt tests --- .../sqllogictest/test_files/spark/math/ceil.slt | 14 ++++++++++---- .../sqllogictest/test_files/spark/math/ceiling.slt | 14 ++++++++++---- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/datafusion/sqllogictest/test_files/spark/math/ceil.slt b/datafusion/sqllogictest/test_files/spark/math/ceil.slt index b7cf476fad7c3..2335e85c01c98 100644 --- a/datafusion/sqllogictest/test_files/spark/math/ceil.slt +++ b/datafusion/sqllogictest/test_files/spark/math/ceil.slt @@ -23,23 +23,29 @@ ## Original Query: SELECT ceil(-0.1); ## PySpark 3.5.5 Result: {'CEIL(-0.1)': Decimal('0'), 'typeof(CEIL(-0.1))': 'decimal(1,0)', 'typeof(-0.1)': 'decimal(1,1)'} -#query -#SELECT ceil(-0.1::decimal(1,1)); +query R +SELECT ceil(-0.1::decimal(1,1)); +---- +0 ## Original Query: SELECT ceil(3.1411, -3); ## PySpark 3.5.5 Result: {'ceil(3.1411, -3)': Decimal('1000'), 'typeof(ceil(3.1411, -3))': 'decimal(4,0)', 'typeof(3.1411)': 'decimal(5,4)', 'typeof(-3)': 'int'} +## TODO: 2-argument ceil(value, scale) is not yet implemented #query #SELECT ceil(3.1411::decimal(5,4), -3::int); ## Original Query: SELECT ceil(3.1411, 3); ## PySpark 3.5.5 Result: {'ceil(3.1411, 3)': Decimal('3.142'), 'typeof(ceil(3.1411, 3))': 'decimal(5,3)', 'typeof(3.1411)': 'decimal(5,4)', 'typeof(3)': 'int'} +## TODO: 2-argument ceil(value, scale) is not yet implemented #query #SELECT ceil(3.1411::decimal(5,4), 3::int); ## Original Query: SELECT ceil(5); ## PySpark 3.5.5 Result: {'CEIL(5)': 5, 'typeof(CEIL(5))': 'bigint', 'typeof(5)': 'int'} -#query -#SELECT ceil(5::int); +query I +SELECT ceil(5::int); +---- +5 # Scalar input: float64 returns bigint query IIIIIII diff --git a/datafusion/sqllogictest/test_files/spark/math/ceiling.slt b/datafusion/sqllogictest/test_files/spark/math/ceiling.slt index b68a4be3092de..8dc05c8d48cca 100644 --- a/datafusion/sqllogictest/test_files/spark/math/ceiling.slt +++ b/datafusion/sqllogictest/test_files/spark/math/ceiling.slt @@ -23,23 +23,29 @@ ## Original Query: SELECT ceiling(-0.1); ## PySpark 3.5.5 Result: {'ceiling(-0.1)': Decimal('0'), 'typeof(ceiling(-0.1))': 'decimal(1,0)', 'typeof(-0.1)': 'decimal(1,1)'} -#query -#SELECT ceiling(-0.1::decimal(1,1)); +query R +SELECT ceiling(-0.1::decimal(1,1)); +---- +0 ## Original Query: SELECT ceiling(3.1411, -3); ## PySpark 3.5.5 Result: {'ceiling(3.1411, -3)': Decimal('1000'), 'typeof(ceiling(3.1411, -3))': 'decimal(4,0)', 'typeof(3.1411)': 'decimal(5,4)', 'typeof(-3)': 'int'} +## TODO: 2-argument ceiling(value, scale) is not yet implemented #query #SELECT ceiling(3.1411::decimal(5,4), -3::int); ## Original Query: SELECT ceiling(3.1411, 3); ## PySpark 3.5.5 Result: {'ceiling(3.1411, 3)': Decimal('3.142'), 'typeof(ceiling(3.1411, 3))': 'decimal(5,3)', 'typeof(3.1411)': 'decimal(5,4)', 'typeof(3)': 'int'} +## TODO: 2-argument ceiling(value, scale) is not yet implemented #query #SELECT ceiling(3.1411::decimal(5,4), 3::int); ## Original Query: SELECT ceiling(5); ## PySpark 3.5.5 Result: {'ceiling(5)': 5, 'typeof(ceiling(5))': 'bigint', 'typeof(5)': 'int'} -#query -#SELECT ceiling(5::int); +query I +SELECT ceiling(5::int); +---- +5 # ceiling is an alias for ceil query I From 0ad51f985953cf08347df7d8126d2b850d323a4a Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Sat, 28 Feb 2026 09:37:16 +0000 Subject: [PATCH 06/25] Expand on diff comment in ceil.rs --- datafusion/spark/src/function/math/ceil.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 3c025e8171479..41fbd7ab5cf95 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -31,8 +31,12 @@ use datafusion_expr::{ /// /// /// Differences with DataFusion ceil: -/// - Spark's ceil returns Int64 for float/integer types -/// - Spark's ceil adjusts precision for Decimal128 types +/// - Spark's ceil returns Int64 for float and integer inputs; DataFusion preserves +/// the input type (Float32→Float32, Float64→Float64, integers coerced to Float64) +/// - Spark's ceil on Decimal128(p, s) returns Decimal128(p−s+1, 0), reducing scale +/// to 0; DataFusion preserves the original precision and scale +/// - Spark only supports Decimal128; DataFusion also supports Decimal32/64/256 +/// - Spark does not check for decimal overflow; DataFusion errors on overflow #[derive(Debug, PartialEq, Eq, Hash)] pub struct SparkCeil { signature: Signature, From 9e7a0fd7fdda9658e3a837b26e5f7b6b5077da26 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 10 Mar 2026 14:07:58 +0000 Subject: [PATCH 07/25] Add comment explaining difference between spark and datafusion in ceil.slt --- .../sqllogictest/test_files/spark/math/ceil.slt | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/datafusion/sqllogictest/test_files/spark/math/ceil.slt b/datafusion/sqllogictest/test_files/spark/math/ceil.slt index 2335e85c01c98..9c8938be5becd 100644 --- a/datafusion/sqllogictest/test_files/spark/math/ceil.slt +++ b/datafusion/sqllogictest/test_files/spark/math/ceil.slt @@ -21,6 +21,17 @@ # For more information, please see: # https://github.com/apache/datafusion/issues/15914 +# Tests for Spark-compatible ceil function. +# Spark semantics differ from DataFusion's built-in ceil in two ways: +# 1. Return type: Spark returns Int64 for float/integer inputs; +# DataFusion returns the same float type (e.g. ceil(1.5::DOUBLE) -> DOUBLE in DF, BIGINT in Spark) +# 2. Decimal precision: Spark adjusts precision to (p - s + 1) for Decimal128(p, s) with scale > 0; +# DataFusion preserves the original precision and scale +# +# Example: SELECT ceil(1.50::DECIMAL(10,2)) +# Spark: returns Decimal(9, 0) value 2 +# DataFusion: returns Decimal(10, 2) value 2.00 + ## Original Query: SELECT ceil(-0.1); ## PySpark 3.5.5 Result: {'CEIL(-0.1)': Decimal('0'), 'typeof(CEIL(-0.1))': 'decimal(1,0)', 'typeof(-0.1)': 'decimal(1,1)'} query R From ab370901d65d47ead3dd0430042789d537e917d7 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 13:04:49 +0000 Subject: [PATCH 08/25] Move if statement inside match --- datafusion/spark/src/function/math/ceil.rs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 41fbd7ab5cf95..0362ff367e6b6 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -73,11 +73,14 @@ impl ScalarUDFImpl for SparkCeil { fn return_type(&self, arg_types: &[DataType]) -> Result { match &arg_types[0] { - DataType::Decimal128(p, s) if *s > 0 => { - let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; - Ok(DataType::Decimal128(new_p, 0)) + DataType::Decimal128(p, s) => { + if *s > 0 { + let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; + Ok(DataType::Decimal128(new_p, 0)) + } else { + Ok(DataType::Decimal128(*p, *s)) + } } - DataType::Decimal128(p, s) => Ok(DataType::Decimal128(*p, *s)), _ => Ok(DataType::Int64), } } From 14b65302f3403600481eb3e1bee4dbfa803465b6 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 13:23:46 +0000 Subject: [PATCH 09/25] Allow any integet input type to remain unchanged, remove cast to Int64 --- datafusion/spark/src/function/math/ceil.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 0362ff367e6b6..aaa7572319c0b 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -19,7 +19,6 @@ use std::any::Any; use std::sync::Arc; use arrow::array::{AsArray, Decimal128Array}; -use arrow::compute::cast; use arrow::datatypes::{DataType, Decimal128Type, Float32Type, Float64Type, Int64Type}; use datafusion_common::utils::take_function_args; use datafusion_common::{Result, exec_err}; @@ -31,8 +30,8 @@ use datafusion_expr::{ /// /// /// Differences with DataFusion ceil: -/// - Spark's ceil returns Int64 for float and integer inputs; DataFusion preserves -/// the input type (Float32→Float32, Float64→Float64, integers coerced to Float64) +/// - Spark's ceil returns Int64 for float inputs; DataFusion preserves +/// the input type (Float32→Float32, Float64→Float64) /// - Spark's ceil on Decimal128(p, s) returns Decimal128(p−s+1, 0), reducing scale /// to 0; DataFusion preserves the original precision and scale /// - Spark only supports Decimal128; DataFusion also supports Decimal32/64/256 @@ -81,6 +80,7 @@ impl ScalarUDFImpl for SparkCeil { Ok(DataType::Decimal128(*p, *s)) } } + dt if dt.is_integer() => Ok(dt.clone()), _ => Ok(DataType::Int64), } } @@ -112,7 +112,7 @@ fn spark_ceil(args: &[ColumnarValue], return_type: &DataType) -> Result() .unary::<_, Int64Type>(|x| x.ceil() as i64), ) as _, - dt if dt.is_integer() => cast(&input, &DataType::Int64)?, + dt if dt.is_integer() => input, DataType::Decimal128(_, s) if *s > 0 => { let div = 10_i128.pow(*s as u32); let result: Decimal128Array = From f3069d1bc6aec6be6aaa0818f8be9ef0dd7999ed Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 13:38:59 +0000 Subject: [PATCH 10/25] Remove unnecessary clone for new decimal datatype --- datafusion/spark/src/function/math/ceil.rs | 24 +++++++++++----------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index aaa7572319c0b..a0cf142904083 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -86,8 +86,7 @@ impl ScalarUDFImpl for SparkCeil { } fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { - let return_type = args.return_type().clone(); - spark_ceil(&args.args, &return_type) + spark_ceil(&args.args) } fn aliases(&self) -> &[String] { @@ -95,7 +94,7 @@ impl ScalarUDFImpl for SparkCeil { } } -fn spark_ceil(args: &[ColumnarValue], return_type: &DataType) -> Result { +fn spark_ceil(args: &[ColumnarValue]) -> Result { let input = match take_function_args("ceil", args)? { [ColumnarValue::Scalar(value)] => value.to_array()?, [ColumnarValue::Array(arr)] => Arc::clone(arr), @@ -113,15 +112,16 @@ fn spark_ceil(args: &[ColumnarValue], return_type: &DataType) -> Result(|x| x.ceil() as i64), ) as _, dt if dt.is_integer() => input, - DataType::Decimal128(_, s) if *s > 0 => { + DataType::Decimal128(p, s) if *s > 0 => { let div = 10_i128.pow(*s as u32); + let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; let result: Decimal128Array = input.as_primitive::().unary(|x| { let d = x / div; let r = x % div; if r > 0 { d + 1 } else { d } }); - Arc::new(result.with_data_type(return_type.clone())) + Arc::new(result.with_data_type(DataType::Decimal128(new_p, 0))) } DataType::Decimal128(_, _) => input, other => return exec_err!("Unsupported data type {other:?} for function ceil"), @@ -148,7 +148,7 @@ mod tests { None, ]); let args = vec![ColumnarValue::Array(Arc::new(input))]; - let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = spark_ceil(&args).unwrap(); let result = match result { ColumnarValue::Array(arr) => arr, _ => panic!("Expected array"), @@ -180,7 +180,7 @@ mod tests { None, ]); let args = vec![ColumnarValue::Array(Arc::new(input))]; - let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = spark_ceil(&args).unwrap(); let result = match result { ColumnarValue::Array(arr) => arr, _ => panic!("Expected array"), @@ -204,7 +204,7 @@ mod tests { fn test_ceil_int64() { let input = Int64Array::from(vec![Some(1), Some(-1), None]); let args = vec![ColumnarValue::Array(Arc::new(input))]; - let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = spark_ceil(&args).unwrap(); let result = match result { ColumnarValue::Array(arr) => arr, _ => panic!("Expected array"), @@ -220,7 +220,7 @@ mod tests { let input = Decimal128Array::from(vec![Some(150), Some(-150), Some(100), None]) .with_data_type(DataType::Decimal128(10, 2)); let args = vec![ColumnarValue::Array(Arc::new(input))]; - let result = spark_ceil(&args, &return_type).unwrap(); + let result = spark_ceil(&args).unwrap(); let result = match result { ColumnarValue::Array(arr) => arr, _ => panic!("Expected array"), @@ -235,7 +235,7 @@ mod tests { fn test_ceil_float64_scalar() { let input = ScalarValue::Float64(Some(-1.1)); let args = vec![ColumnarValue::Scalar(input)]; - let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = spark_ceil(&args).unwrap(); let result = match result { ColumnarValue::Array(arr) => arr, _ => panic!("Expected array"), @@ -248,7 +248,7 @@ mod tests { fn test_ceil_float32_scalar() { let input = ScalarValue::Float32(Some(125.2345f32)); let args = vec![ColumnarValue::Scalar(input)]; - let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = spark_ceil(&args).unwrap(); let result = match result { ColumnarValue::Array(arr) => arr, _ => panic!("Expected array"), @@ -261,7 +261,7 @@ mod tests { fn test_ceil_int64_scalar() { let input = ScalarValue::Int64(Some(48)); let args = vec![ColumnarValue::Scalar(input)]; - let result = spark_ceil(&args, &DataType::Int64).unwrap(); + let result = spark_ceil(&args).unwrap(); let result = match result { ColumnarValue::Array(arr) => arr, _ => panic!("Expected array"), From ce9f3455efd9d15989895ef304fa8513ddd659ad Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 13:48:27 +0000 Subject: [PATCH 11/25] Handle scalars and arrays separately --- datafusion/spark/src/function/math/ceil.rs | 72 ++++++++++++++-------- 1 file changed, 48 insertions(+), 24 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index a0cf142904083..21a67b42d4557 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -21,7 +21,7 @@ use std::sync::Arc; use arrow::array::{AsArray, Decimal128Array}; use arrow::datatypes::{DataType, Decimal128Type, Float32Type, Float64Type, Int64Type}; use datafusion_common::utils::take_function_args; -use datafusion_common::{Result, exec_err}; +use datafusion_common::{Result, ScalarValue, exec_err}; use datafusion_expr::{ ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility, }; @@ -95,11 +95,41 @@ impl ScalarUDFImpl for SparkCeil { } fn spark_ceil(args: &[ColumnarValue]) -> Result { - let input = match take_function_args("ceil", args)? { - [ColumnarValue::Scalar(value)] => value.to_array()?, - [ColumnarValue::Array(arr)] => Arc::clone(arr), + let [input] = take_function_args("ceil", args)?; + + match input { + ColumnarValue::Scalar(value) => spark_ceil_scalar(value), + ColumnarValue::Array(input) => spark_ceil_array(input), + } +} + +fn spark_ceil_scalar(value: &ScalarValue) -> Result { + let result = match value { + ScalarValue::Float32(v) => ScalarValue::Int64(v.map(|x| x.ceil() as i64)), + ScalarValue::Float64(v) => ScalarValue::Int64(v.map(|x| x.ceil() as i64)), + v if v.data_type().is_integer() => v.clone(), + ScalarValue::Decimal128(v, p, s) if *s > 0 => { + let div = 10_i128.pow(*s as u32); + let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; + let result = v.map(|x| { + let d = x / div; + let r = x % div; + if r > 0 { d + 1 } else { d } + }); + ScalarValue::Decimal128(result, new_p, 0) + } + ScalarValue::Decimal128(_, _, _) => value.clone(), + other => { + return exec_err!( + "Unsupported data type {:?} for function ceil", + other.data_type() + ); + } }; + Ok(ColumnarValue::Scalar(result)) +} +fn spark_ceil_array(input: &Arc) -> Result { let result = match input.data_type() { DataType::Float32 => Arc::new( input @@ -111,7 +141,7 @@ fn spark_ceil(args: &[ColumnarValue]) -> Result { .as_primitive::() .unary::<_, Int64Type>(|x| x.ceil() as i64), ) as _, - dt if dt.is_integer() => input, + dt if dt.is_integer() => Arc::clone(input), DataType::Decimal128(p, s) if *s > 0 => { let div = 10_i128.pow(*s as u32); let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; @@ -123,7 +153,7 @@ fn spark_ceil(args: &[ColumnarValue]) -> Result { }); Arc::new(result.with_data_type(DataType::Decimal128(new_p, 0))) } - DataType::Decimal128(_, _) => input, + DataType::Decimal128(_, _) => Arc::clone(input), other => return exec_err!("Unsupported data type {other:?} for function ceil"), }; @@ -235,38 +265,32 @@ mod tests { fn test_ceil_float64_scalar() { let input = ScalarValue::Float64(Some(-1.1)); let args = vec![ColumnarValue::Scalar(input)]; - let result = spark_ceil(&args).unwrap(); - let result = match result { - ColumnarValue::Array(arr) => arr, - _ => panic!("Expected array"), + let result = match spark_ceil(&args).unwrap() { + ColumnarValue::Scalar(v) => v, + _ => panic!("Expected scalar"), }; - let result = result.as_primitive::(); - assert_eq!(result, &Int64Array::from(vec![Some(-1)])); + assert_eq!(result, ScalarValue::Int64(Some(-1))); } #[test] fn test_ceil_float32_scalar() { let input = ScalarValue::Float32(Some(125.2345f32)); let args = vec![ColumnarValue::Scalar(input)]; - let result = spark_ceil(&args).unwrap(); - let result = match result { - ColumnarValue::Array(arr) => arr, - _ => panic!("Expected array"), + let result = match spark_ceil(&args).unwrap() { + ColumnarValue::Scalar(v) => v, + _ => panic!("Expected scalar"), }; - let result = result.as_primitive::(); - assert_eq!(result, &Int64Array::from(vec![Some(126)])); + assert_eq!(result, ScalarValue::Int64(Some(126))); } #[test] fn test_ceil_int64_scalar() { let input = ScalarValue::Int64(Some(48)); let args = vec![ColumnarValue::Scalar(input)]; - let result = spark_ceil(&args).unwrap(); - let result = match result { - ColumnarValue::Array(arr) => arr, - _ => panic!("Expected array"), + let result = match spark_ceil(&args).unwrap() { + ColumnarValue::Scalar(v) => v, + _ => panic!("Expected scalar"), }; - let result = result.as_primitive::(); - assert_eq!(result, &Int64Array::from(vec![Some(48)])); + assert_eq!(result, ScalarValue::Int64(Some(48))); } } From 336870f5a0069bad6ff499a34fd95eaa6edb3a86 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 13:54:30 +0000 Subject: [PATCH 12/25] No ceiling alias --- datafusion/spark/src/function/math/ceil.rs | 6 ------ 1 file changed, 6 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 21a67b42d4557..0a6c9bb70e5a1 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -39,7 +39,6 @@ use datafusion_expr::{ #[derive(Debug, PartialEq, Eq, Hash)] pub struct SparkCeil { signature: Signature, - aliases: Vec, } impl Default for SparkCeil { @@ -52,7 +51,6 @@ impl SparkCeil { pub fn new() -> Self { Self { signature: Signature::numeric(1, Volatility::Immutable), - aliases: vec!["ceiling".to_string()], } } } @@ -88,10 +86,6 @@ impl ScalarUDFImpl for SparkCeil { fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { spark_ceil(&args.args) } - - fn aliases(&self) -> &[String] { - &self.aliases - } } fn spark_ceil(args: &[ColumnarValue]) -> Result { From 3dce8df5bbe788f3cb391f6184e4627e2fc26e46 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 13:55:40 +0000 Subject: [PATCH 13/25] Add comment explaining why negative scale doesn't affect datatype --- datafusion/spark/src/function/math/ceil.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 0a6c9bb70e5a1..fdf9ca88476a8 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -75,6 +75,8 @@ impl ScalarUDFImpl for SparkCeil { let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; Ok(DataType::Decimal128(new_p, 0)) } else { + // scale <= 0 means the value is already a whole number + // (or represents multiples of 10^(-scale)), so ceil is a no-op Ok(DataType::Decimal128(*p, *s)) } } From f2c60203aa604c88d308585434d1e7b3ccf1f2a2 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 13:57:45 +0000 Subject: [PATCH 14/25] Exhaustive match for return_type from input datatype --- datafusion/spark/src/function/math/ceil.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index fdf9ca88476a8..ccaa4ffa2f4ba 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -81,7 +81,8 @@ impl ScalarUDFImpl for SparkCeil { } } dt if dt.is_integer() => Ok(dt.clone()), - _ => Ok(DataType::Int64), + DataType::Float32 | DataType::Float64 => Ok(DataType::Int64), + other => exec_err!("Unsupported data type {other:?} for function ceil"), } } From 8d4c63f00dbbc5e5026d644bdfcae077e78ccfaf Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 13:59:10 +0000 Subject: [PATCH 15/25] Add TODO note --- datafusion/spark/src/function/math/ceil.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index ccaa4ffa2f4ba..a6c097d4aa32a 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -36,6 +36,8 @@ use datafusion_expr::{ /// to 0; DataFusion preserves the original precision and scale /// - Spark only supports Decimal128; DataFusion also supports Decimal32/64/256 /// - Spark does not check for decimal overflow; DataFusion errors on overflow +/// +/// TODO: 2-argument ceil(value, scale) is not yet implemented #[derive(Debug, PartialEq, Eq, Hash)] pub struct SparkCeil { signature: Signature, From 2d3658f5e27178ca88b9a12ba02f98e0bc0c94f3 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 14:00:46 +0000 Subject: [PATCH 16/25] Delete old ceiling.slt since we removed the alias --- .../test_files/spark/math/ceiling.slt | 64 ------------------- 1 file changed, 64 deletions(-) delete mode 100644 datafusion/sqllogictest/test_files/spark/math/ceiling.slt diff --git a/datafusion/sqllogictest/test_files/spark/math/ceiling.slt b/datafusion/sqllogictest/test_files/spark/math/ceiling.slt deleted file mode 100644 index 8dc05c8d48cca..0000000000000 --- a/datafusion/sqllogictest/test_files/spark/math/ceiling.slt +++ /dev/null @@ -1,64 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at - -# http://www.apache.org/licenses/LICENSE-2.0 - -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -# This file was originally created by a porting script from: -# https://github.com/lakehq/sail/tree/43b6ed8221de5c4c4adbedbb267ae1351158b43c/crates/sail-spark-connect/tests/gold_data/function -# This file is part of the implementation of the datafusion-spark function library. -# For more information, please see: -# https://github.com/apache/datafusion/issues/15914 - -## Original Query: SELECT ceiling(-0.1); -## PySpark 3.5.5 Result: {'ceiling(-0.1)': Decimal('0'), 'typeof(ceiling(-0.1))': 'decimal(1,0)', 'typeof(-0.1)': 'decimal(1,1)'} -query R -SELECT ceiling(-0.1::decimal(1,1)); ----- -0 - -## Original Query: SELECT ceiling(3.1411, -3); -## PySpark 3.5.5 Result: {'ceiling(3.1411, -3)': Decimal('1000'), 'typeof(ceiling(3.1411, -3))': 'decimal(4,0)', 'typeof(3.1411)': 'decimal(5,4)', 'typeof(-3)': 'int'} -## TODO: 2-argument ceiling(value, scale) is not yet implemented -#query -#SELECT ceiling(3.1411::decimal(5,4), -3::int); - -## Original Query: SELECT ceiling(3.1411, 3); -## PySpark 3.5.5 Result: {'ceiling(3.1411, 3)': Decimal('3.142'), 'typeof(ceiling(3.1411, 3))': 'decimal(5,3)', 'typeof(3.1411)': 'decimal(5,4)', 'typeof(3)': 'int'} -## TODO: 2-argument ceiling(value, scale) is not yet implemented -#query -#SELECT ceiling(3.1411::decimal(5,4), 3::int); - -## Original Query: SELECT ceiling(5); -## PySpark 3.5.5 Result: {'ceiling(5)': 5, 'typeof(ceiling(5))': 'bigint', 'typeof(5)': 'int'} -query I -SELECT ceiling(5::int); ----- -5 - -# ceiling is an alias for ceil -query I -SELECT ceiling(1.5::DOUBLE); ----- -2 - -query I -SELECT ceiling(5::INT); ----- -5 - -query R -SELECT ceiling(1.50::DECIMAL(10, 2)); ----- -2 From fd23001a99d5479955e91bad9e411fd790f3f345 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 14:08:05 +0000 Subject: [PATCH 17/25] Better wording in mod.rs --- datafusion/spark/src/function/math/mod.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/datafusion/spark/src/function/math/mod.rs b/datafusion/spark/src/function/math/mod.rs index c14d6cdf50606..dc2b136b4e91a 100644 --- a/datafusion/spark/src/function/math/mod.rs +++ b/datafusion/spark/src/function/math/mod.rs @@ -51,11 +51,7 @@ pub mod expr_fn { use datafusion_functions::export_functions; export_functions!((abs, "Returns abs(expr)", arg1)); - export_functions!(( - ceil, - "Returns the smallest integer not less than expr.", - arg1 - )); + export_functions!((ceil, "Returns the ceiling of expr.", arg1)); export_functions!((expm1, "Returns exp(expr) - 1 as a Float64.", arg1)); export_functions!(( factorial, From 1b04c3f40577c8016740de830b8d707149f3f59b Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 17 Mar 2026 14:17:10 +0000 Subject: [PATCH 18/25] Pow wrapping instead --- datafusion/spark/src/function/math/ceil.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index a6c097d4aa32a..2757ec835a88f 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -18,7 +18,7 @@ use std::any::Any; use std::sync::Arc; -use arrow::array::{AsArray, Decimal128Array}; +use arrow::array::{ArrowNativeTypeOp, AsArray, Decimal128Array}; use arrow::datatypes::{DataType, Decimal128Type, Float32Type, Float64Type, Int64Type}; use datafusion_common::utils::take_function_args; use datafusion_common::{Result, ScalarValue, exec_err}; @@ -108,7 +108,7 @@ fn spark_ceil_scalar(value: &ScalarValue) -> Result { ScalarValue::Float64(v) => ScalarValue::Int64(v.map(|x| x.ceil() as i64)), v if v.data_type().is_integer() => v.clone(), ScalarValue::Decimal128(v, p, s) if *s > 0 => { - let div = 10_i128.pow(*s as u32); + let div = 10_i128.pow_wrapping(*s as u32); let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; let result = v.map(|x| { let d = x / div; @@ -142,7 +142,7 @@ fn spark_ceil_array(input: &Arc) -> Result Arc::clone(input), DataType::Decimal128(p, s) if *s > 0 => { - let div = 10_i128.pow(*s as u32); + let div = 10_i128.pow_wrapping(*s as u32); let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; let result: Decimal128Array = input.as_primitive::().unary(|x| { From 5ad7ef6d710a57e9865415945ab4891709a64b45 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 7 Apr 2026 13:59:40 +0100 Subject: [PATCH 19/25] Make return types match spark, always int64 except for decimal input --- datafusion/spark/src/function/math/ceil.rs | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 2757ec835a88f..cd7ddfdf0648f 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -15,7 +15,6 @@ // specific language governing permissions and limitations // under the License. -use std::any::Any; use std::sync::Arc; use arrow::array::{ArrowNativeTypeOp, AsArray, Decimal128Array}; @@ -58,10 +57,6 @@ impl SparkCeil { } impl ScalarUDFImpl for SparkCeil { - fn as_any(&self) -> &dyn Any { - self - } - fn name(&self) -> &str { "ceil" } @@ -82,8 +77,8 @@ impl ScalarUDFImpl for SparkCeil { Ok(DataType::Decimal128(*p, *s)) } } - dt if dt.is_integer() => Ok(dt.clone()), DataType::Float32 | DataType::Float64 => Ok(DataType::Int64), + dt if dt.is_integer() => Ok(DataType::Int64), other => exec_err!("Unsupported data type {other:?} for function ceil"), } } @@ -106,7 +101,7 @@ fn spark_ceil_scalar(value: &ScalarValue) -> Result { let result = match value { ScalarValue::Float32(v) => ScalarValue::Int64(v.map(|x| x.ceil() as i64)), ScalarValue::Float64(v) => ScalarValue::Int64(v.map(|x| x.ceil() as i64)), - v if v.data_type().is_integer() => v.clone(), + v if v.data_type().is_integer() => v.cast_to(&DataType::Int64)?, ScalarValue::Decimal128(v, p, s) if *s > 0 => { let div = 10_i128.pow_wrapping(*s as u32); let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; @@ -140,7 +135,7 @@ fn spark_ceil_array(input: &Arc) -> Result() .unary::<_, Int64Type>(|x| x.ceil() as i64), ) as _, - dt if dt.is_integer() => Arc::clone(input), + dt if dt.is_integer() => arrow::compute::cast(input, &DataType::Int64)?, DataType::Decimal128(p, s) if *s > 0 => { let div = 10_i128.pow_wrapping(*s as u32); let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; From 9b2331830bbaabfc7ba1b2aec3b5880f8aa8087a Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 7 Apr 2026 14:00:54 +0100 Subject: [PATCH 20/25] Reduce duplication between scalar and array functions --- datafusion/spark/src/function/math/ceil.rs | 40 ++++++++++++---------- 1 file changed, 22 insertions(+), 18 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index cd7ddfdf0648f..6991b447ec812 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -69,8 +69,7 @@ impl ScalarUDFImpl for SparkCeil { match &arg_types[0] { DataType::Decimal128(p, s) => { if *s > 0 { - let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; - Ok(DataType::Decimal128(new_p, 0)) + Ok(DataType::Decimal128(decimal128_ceil_precision(*p, *s), 0)) } else { // scale <= 0 means the value is already a whole number // (or represents multiples of 10^(-scale)), so ceil is a no-op @@ -97,20 +96,29 @@ fn spark_ceil(args: &[ColumnarValue]) -> Result { } } +/// Compute ceil for a single decimal128 value with the given scale. +#[inline] +fn decimal128_ceil(value: i128, scale: u32) -> i128 { + let div = 10_i128.pow_wrapping(scale); + let d = value / div; + let r = value % div; + if r > 0 { d + 1 } else { d } +} + +/// Compute the return precision for a decimal128 ceil result. +#[inline] +fn decimal128_ceil_precision(precision: u8, scale: i8) -> u8 { + ((precision as i64) - (scale as i64) + 1).clamp(1, 38) as u8 +} + fn spark_ceil_scalar(value: &ScalarValue) -> Result { let result = match value { ScalarValue::Float32(v) => ScalarValue::Int64(v.map(|x| x.ceil() as i64)), ScalarValue::Float64(v) => ScalarValue::Int64(v.map(|x| x.ceil() as i64)), v if v.data_type().is_integer() => v.cast_to(&DataType::Int64)?, ScalarValue::Decimal128(v, p, s) if *s > 0 => { - let div = 10_i128.pow_wrapping(*s as u32); - let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; - let result = v.map(|x| { - let d = x / div; - let r = x % div; - if r > 0 { d + 1 } else { d } - }); - ScalarValue::Decimal128(result, new_p, 0) + let new_p = decimal128_ceil_precision(*p, *s); + ScalarValue::Decimal128(v.map(|x| decimal128_ceil(x, *s as u32)), new_p, 0) } ScalarValue::Decimal128(_, _, _) => value.clone(), other => { @@ -137,14 +145,10 @@ fn spark_ceil_array(input: &Arc) -> Result arrow::compute::cast(input, &DataType::Int64)?, DataType::Decimal128(p, s) if *s > 0 => { - let div = 10_i128.pow_wrapping(*s as u32); - let new_p = ((*p as i64) - (*s as i64) + 1).clamp(1, 38) as u8; - let result: Decimal128Array = - input.as_primitive::().unary(|x| { - let d = x / div; - let r = x % div; - if r > 0 { d + 1 } else { d } - }); + let new_p = decimal128_ceil_precision(*p, *s); + let result: Decimal128Array = input + .as_primitive::() + .unary(|x| decimal128_ceil(x, *s as u32)); Arc::new(result.with_data_type(DataType::Decimal128(new_p, 0))) } DataType::Decimal128(_, _) => Arc::clone(input), From e839b6ef0ba77d9bedb5bb0e4866ca4242b086da Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Tue, 7 Apr 2026 14:05:48 +0100 Subject: [PATCH 21/25] Add type assertions to slt test --- .../test_files/spark/math/ceil.slt | 72 +++++++++---------- 1 file changed, 36 insertions(+), 36 deletions(-) diff --git a/datafusion/sqllogictest/test_files/spark/math/ceil.slt b/datafusion/sqllogictest/test_files/spark/math/ceil.slt index 9c8938be5becd..087acf000b6f1 100644 --- a/datafusion/sqllogictest/test_files/spark/math/ceil.slt +++ b/datafusion/sqllogictest/test_files/spark/math/ceil.slt @@ -34,10 +34,10 @@ ## Original Query: SELECT ceil(-0.1); ## PySpark 3.5.5 Result: {'CEIL(-0.1)': Decimal('0'), 'typeof(CEIL(-0.1))': 'decimal(1,0)', 'typeof(-0.1)': 'decimal(1,1)'} -query R -SELECT ceil(-0.1::decimal(1,1)); +query RT +SELECT ceil(-0.1::decimal(1,1)), arrow_typeof(ceil(-0.1::decimal(1,1))); ---- -0 +0 Decimal128(1, 0) ## Original Query: SELECT ceil(3.1411, -3); ## PySpark 3.5.5 Result: {'ceil(3.1411, -3)': Decimal('1000'), 'typeof(ceil(3.1411, -3))': 'decimal(4,0)', 'typeof(3.1411)': 'decimal(5,4)', 'typeof(-3)': 'int'} @@ -53,68 +53,68 @@ SELECT ceil(-0.1::decimal(1,1)); ## Original Query: SELECT ceil(5); ## PySpark 3.5.5 Result: {'CEIL(5)': 5, 'typeof(CEIL(5))': 'bigint', 'typeof(5)': 'int'} -query I -SELECT ceil(5::int); +query IT +SELECT ceil(5::int), arrow_typeof(ceil(5::int)); ---- -5 +5 Int64 # Scalar input: float64 returns bigint -query IIIIIII -SELECT ceil(125.2345::DOUBLE), ceil(15.0001::DOUBLE), ceil(0.1::DOUBLE), ceil(-0.9::DOUBLE), ceil(-1.1::DOUBLE), ceil(123.0::DOUBLE), ceil(NULL::DOUBLE); +query IIIIIIIT +SELECT ceil(125.2345::DOUBLE), ceil(15.0001::DOUBLE), ceil(0.1::DOUBLE), ceil(-0.9::DOUBLE), ceil(-1.1::DOUBLE), ceil(123.0::DOUBLE), ceil(NULL::DOUBLE), arrow_typeof(ceil(125.2345::DOUBLE)); ---- -126 16 1 0 -1 123 NULL +126 16 1 0 -1 123 NULL Int64 # Scalar input: float32 returns bigint -query IIIIIII -SELECT ceil(125.2345::FLOAT), ceil(15.0001::FLOAT), ceil(0.1::FLOAT), ceil(-0.9::FLOAT), ceil(-1.1::FLOAT), ceil(123.0::FLOAT), ceil(NULL::FLOAT); +query IIIIIIIT +SELECT ceil(125.2345::FLOAT), ceil(15.0001::FLOAT), ceil(0.1::FLOAT), ceil(-0.9::FLOAT), ceil(-1.1::FLOAT), ceil(123.0::FLOAT), ceil(NULL::FLOAT), arrow_typeof(ceil(125.2345::FLOAT)); ---- -126 16 1 0 -1 123 NULL +126 16 1 0 -1 123 NULL Int64 # Scalar input: integer types all return bigint -query III -SELECT ceil(5::TINYINT), ceil(-3::TINYINT), ceil(NULL::TINYINT); +query IIIT +SELECT ceil(5::TINYINT), ceil(-3::TINYINT), ceil(NULL::TINYINT), arrow_typeof(ceil(5::TINYINT)); ---- -5 -3 NULL +5 -3 NULL Int64 -query III -SELECT ceil(5::SMALLINT), ceil(-3::SMALLINT), ceil(NULL::SMALLINT); +query IIIT +SELECT ceil(5::SMALLINT), ceil(-3::SMALLINT), ceil(NULL::SMALLINT), arrow_typeof(ceil(5::SMALLINT)); ---- -5 -3 NULL +5 -3 NULL Int64 -query III -SELECT ceil(5::INT), ceil(-3::INT), ceil(NULL::INT); +query IIIT +SELECT ceil(5::INT), ceil(-3::INT), ceil(NULL::INT), arrow_typeof(ceil(5::INT)); ---- -5 -3 NULL +5 -3 NULL Int64 -query III -SELECT ceil(5::BIGINT), ceil(-3::BIGINT), ceil(NULL::BIGINT); +query IIIT +SELECT ceil(5::BIGINT), ceil(-3::BIGINT), ceil(NULL::BIGINT), arrow_typeof(ceil(5::BIGINT)); ---- -5 -3 NULL +5 -3 NULL Int64 # Scalar input: decimal128 with scale > 0 returns decimal with scale 0 # ceil(1.50) = 2, ceil(-1.50) = -1, ceil(1.00) = 1 -query RRR -SELECT ceil(1.50::DECIMAL(10, 2)), ceil(-1.50::DECIMAL(10, 2)), ceil(1.00::DECIMAL(10, 2)); +query RRRT +SELECT ceil(1.50::DECIMAL(10, 2)), ceil(-1.50::DECIMAL(10, 2)), ceil(1.00::DECIMAL(10, 2)), arrow_typeof(ceil(1.50::DECIMAL(10, 2))); ---- -2 -1 1 +2 -1 1 Decimal128(9, 0) # ceil(-0.1) = 0 (smallest positive decimal rounds up to 0 for negatives) -query RR -SELECT ceil(-0.1::DECIMAL(3, 1)), ceil(NULL::DECIMAL(10, 2)); +query RRT +SELECT ceil(-0.1::DECIMAL(3, 1)), ceil(NULL::DECIMAL(10, 2)), arrow_typeof(ceil(-0.1::DECIMAL(3, 1))); ---- -0 NULL +0 NULL Decimal128(3, 0) # ceil(3.1411) = 4 -query R -SELECT ceil(3.1411::DECIMAL(5, 4)); +query RT +SELECT ceil(3.1411::DECIMAL(5, 4)), arrow_typeof(ceil(3.1411::DECIMAL(5, 4))); ---- -4 +4 Decimal128(2, 0) # Scalar input: decimal128 with scale = 0 passes through unchanged -query RRR -SELECT ceil(5::DECIMAL(10, 0)), ceil(-3::DECIMAL(10, 0)), ceil(NULL::DECIMAL(10, 0)); +query RRRT +SELECT ceil(5::DECIMAL(10, 0)), ceil(-3::DECIMAL(10, 0)), ceil(NULL::DECIMAL(10, 0)), arrow_typeof(ceil(5::DECIMAL(10, 0))); ---- -5 -3 NULL +5 -3 NULL Decimal128(10, 0) # Array input: float64 query I From 7ec51997f47cd79cc02e087594728b99f4bf24fa Mon Sep 17 00:00:00 2001 From: Oleks V Date: Sat, 11 Apr 2026 12:25:40 -0700 Subject: [PATCH 22/25] Apply suggestion from @comphead --- datafusion/spark/src/function/math/ceil.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 6991b447ec812..28a4e511e0cd7 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -36,7 +36,8 @@ use datafusion_expr::{ /// - Spark only supports Decimal128; DataFusion also supports Decimal32/64/256 /// - Spark does not check for decimal overflow; DataFusion errors on overflow /// -/// TODO: 2-argument ceil(value, scale) is not yet implemented +/// 2-argument ceil(value, scale) is not yet implemented +/// https://github.com/apache/datafusion/issues/21560 #[derive(Debug, PartialEq, Eq, Hash)] pub struct SparkCeil { signature: Signature, From cbe73809c76b94ee846a9f30c46428ca905cebda Mon Sep 17 00:00:00 2001 From: Oleks V Date: Sat, 11 Apr 2026 12:33:21 -0700 Subject: [PATCH 23/25] Apply suggestion from @comphead --- datafusion/spark/src/function/math/ceil.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 28a4e511e0cd7..0f88c99788842 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -36,8 +36,8 @@ use datafusion_expr::{ /// - Spark only supports Decimal128; DataFusion also supports Decimal32/64/256 /// - Spark does not check for decimal overflow; DataFusion errors on overflow /// -/// 2-argument ceil(value, scale) is not yet implemented -/// https://github.com/apache/datafusion/issues/21560 +/// 2-argument ceil(value, scale) is not yet implemented +/// #[derive(Debug, PartialEq, Eq, Hash)] pub struct SparkCeil { signature: Signature, From 174833bb65943d0a4a902f262b695e1e96a24512 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Sat, 11 Apr 2026 20:35:00 +0100 Subject: [PATCH 24/25] Unify match arms --- datafusion/spark/src/function/math/ceil.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 28a4e511e0cd7..3cdf4f9e7ad75 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -77,8 +77,11 @@ impl ScalarUDFImpl for SparkCeil { Ok(DataType::Decimal128(*p, *s)) } } - DataType::Float32 | DataType::Float64 => Ok(DataType::Int64), - dt if dt.is_integer() => Ok(DataType::Int64), + dt if matches!(dt, DataType::Float32 | DataType::Float64) + || dt.is_integer() => + { + Ok(DataType::Int64) + } other => exec_err!("Unsupported data type {other:?} for function ceil"), } } From d76ba2acf3e335630f4c57b4504f482e9061a678 Mon Sep 17 00:00:00 2001 From: Shiv Bhatia Date: Sat, 11 Apr 2026 20:37:50 +0100 Subject: [PATCH 25/25] Add ceiling alias --- datafusion/spark/src/function/math/ceil.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/datafusion/spark/src/function/math/ceil.rs b/datafusion/spark/src/function/math/ceil.rs index 701e370830993..5096914a1eba8 100644 --- a/datafusion/spark/src/function/math/ceil.rs +++ b/datafusion/spark/src/function/math/ceil.rs @@ -41,6 +41,7 @@ use datafusion_expr::{ #[derive(Debug, PartialEq, Eq, Hash)] pub struct SparkCeil { signature: Signature, + aliases: Vec, } impl Default for SparkCeil { @@ -53,6 +54,7 @@ impl SparkCeil { pub fn new() -> Self { Self { signature: Signature::numeric(1, Volatility::Immutable), + aliases: vec!["ceiling".to_string()], } } } @@ -86,6 +88,10 @@ impl ScalarUDFImpl for SparkCeil { } } + fn aliases(&self) -> &[String] { + &self.aliases + } + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { spark_ceil(&args.args) }