diff --git a/datafusion/spark/src/function/math/hypot.rs b/datafusion/spark/src/function/math/hypot.rs new file mode 100644 index 0000000000000..2f88a3fd9559d --- /dev/null +++ b/datafusion/spark/src/function/math/hypot.rs @@ -0,0 +1,90 @@ +// 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::sync::Arc; + +use arrow::array::{ArrayRef, AsArray, Float64Array}; +use arrow::compute::kernels::arity::binary; +use arrow::datatypes::{DataType, Float64Type}; +use datafusion_common::Result; +use datafusion_common::utils::take_function_args; +use datafusion_expr::{ + ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility, +}; + +/// Spark-compatible `hypot` function. +/// +/// +/// +/// Returns `sqrt(expr1^2 + expr2^2)` computed without intermediate overflow or +/// underflow, matching Spark's use of `java.lang.Math.hypot`. +#[derive(Debug, PartialEq, Eq, Hash)] +pub struct SparkHypot { + signature: Signature, +} + +impl Default for SparkHypot { + fn default() -> Self { + Self::new() + } +} + +impl SparkHypot { + pub fn new() -> Self { + Self { + // Spark only defines hypot over doubles; `exact` makes coercion + // guarantee both inputs are Float64 before `invoke` runs. + signature: Signature::exact( + vec![DataType::Float64, DataType::Float64], + Volatility::Immutable, + ), + } + } +} + +impl ScalarUDFImpl for SparkHypot { + fn name(&self) -> &str { + "hypot" + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, _arg_types: &[DataType]) -> Result { + Ok(DataType::Float64) + } + + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { + let num_rows = args.number_rows; + let [x, y] = take_function_args(self.name(), args.args)?; + + // Broadcast scalars to arrays so one path covers every combination. + let x = x.to_array(num_rows)?; + let y = y.to_array(num_rows)?; + + // Safe: the `exact` signature guarantees Float64 inputs. + let x = x.as_primitive::(); + let y = y.as_primitive::(); + + // `binary` applies the op element-wise and returns NULL when either + // input is NULL — matching Spark's null semantics. + let result: Float64Array = binary(x, y, |a, b| a.hypot(b))?; + + Ok(ColumnarValue::Array(Arc::new(result) as ArrayRef)) + } +} diff --git a/datafusion/spark/src/function/math/mod.rs b/datafusion/spark/src/function/math/mod.rs index 0079ef0fc97cd..fb57b536f26ec 100644 --- a/datafusion/spark/src/function/math/mod.rs +++ b/datafusion/spark/src/function/math/mod.rs @@ -22,6 +22,7 @@ pub mod expm1; pub mod factorial; pub mod floor; pub mod hex; +pub mod hypot; pub mod modulus; pub mod negative; pub mod pow; @@ -41,6 +42,7 @@ make_udf_function!(expm1::SparkExpm1, expm1); make_udf_function!(factorial::SparkFactorial, factorial); make_udf_function!(floor::SparkFloor, floor); make_udf_function!(hex::SparkHex, hex); +make_udf_function!(hypot::SparkHypot, hypot); make_udf_function!(modulus::SparkMod, modulus); make_udf_function!(modulus::SparkPmod, pmod); make_udf_function!(pow::SparkPow, pow); @@ -66,6 +68,7 @@ pub mod expr_fn { )); export_functions!((floor, "Returns floor of expr.", arg1)); export_functions!((hex, "Computes hex value of the given column.", arg1)); + export_functions!((hypot, "Returns sqrt(a^2 + b^2) without intermediate overflow or underflow.", arg1 arg2)); export_functions!((modulus, "Returns the remainder of division of the first argument by the second argument.", arg1 arg2)); export_functions!((pmod, "Returns the positive remainder of division of the first argument by the second argument.", arg1 arg2)); export_functions!(( @@ -107,6 +110,7 @@ pub fn functions() -> Vec> { factorial(), floor(), hex(), + hypot(), modulus(), pmod(), pow(), diff --git a/datafusion/sqllogictest/test_files/spark/math/hypot.slt b/datafusion/sqllogictest/test_files/spark/math/hypot.slt index 1349be0a95ee7..3d9d03da7a37e 100644 --- a/datafusion/sqllogictest/test_files/spark/math/hypot.slt +++ b/datafusion/sqllogictest/test_files/spark/math/hypot.slt @@ -21,7 +21,45 @@ # For more information, please see: # https://github.com/apache/datafusion/issues/15914 -## Original Query: SELECT hypot(3, 4); -## PySpark 3.5.5 Result: {'HYPOT(3, 4)': 5.0, 'typeof(HYPOT(3, 4))': 'double', 'typeof(3)': 'int', 'typeof(4)': 'int'} -#query -#SELECT hypot(3::int, 4::int); +# Scalar: classic Pythagorean triples (3-4-5, 5-12-13) +query R +SELECT hypot(3, 4); +---- +5 + +query R +SELECT hypot(5, 12); +---- +13 + +# Double inputs +query R +SELECT hypot(3.0::double, 4.0::double); +---- +5 + +# NULL if either argument is NULL (binary kernel null propagation) +query R +SELECT hypot(NULL::double, 4.0::double); +---- +NULL + +query R +SELECT hypot(3.0::double, NULL::double); +---- +NULL + +# Array path, including a NULL row +query R +SELECT hypot(a, b) FROM (VALUES (3.0::double, 4.0::double), (6.0::double, 8.0::double), (NULL::double, 1.0::double)) AS t(a, b); +---- +5 +10 +NULL + +# Overflow-safe: a naive sqrt(a*a + b*b) would overflow to Infinity here, +# but f64::hypot (like Spark's Math.hypot) stays finite. +query B +SELECT hypot(3e200::double, 4e200::double) < 'Infinity'::double; +---- +true