feat: add Spark-compatible hypot function

datafusion/spark/src/function/math/hypot.rs

@@ -0,0 +1,84 @@
+// 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,
+};
+use datafusion_functions::utils::make_scalar_function;
+
+/// Spark-compatible `hypot` function.
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#hypot>
+///
+/// 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
+            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<DataType> {
+        Ok(DataType::Float64)
+    }
+
+    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+        make_scalar_function(spark_hypot, vec![])(&args.args)
+    }
+}
+
+fn spark_hypot(args: &[ArrayRef]) -> Result<ArrayRef> {
+    let [x, y] = take_function_args("hypot", args)?;
+
+    let x = x.as_primitive::<Float64Type>();
+    let y = y.as_primitive::<Float64Type>();
+    let result: Float64Array = binary(x, y, |a, b| a.hypot(b))?;
+    Ok(Arc::new(result))
+}

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<Arc<ScalarUDF>> {
         factorial(),
         floor(),
         hex(),
+        hypot(),
         modulus(),
         pmod(),
         pow(),

datafusion/sqllogictest/test_files/spark/math/hypot.slt

@@ -21,7 +21,115 @@
 # 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
+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: naive sqrt(a*a + b*b) overflows to Infinity here; hypot stays finite (matches Spark's Math.hypot)
+query B
+SELECT hypot(3e200::double, 4e200::double) < 'Infinity'::double;
+----
+true
+
+# any infinite input yields +Infinity, even when the other is NaN
+query R
+SELECT hypot('Infinity'::double, 4.0::double);
+----
+Infinity
+
+query R
+SELECT hypot(4.0::double, '-Infinity'::double);
+----
+Infinity
+
+query R
+SELECT hypot('Infinity'::double, 'NaN'::double);
+----
+Infinity
+
+# NaN propagates when neither input is infinite
+query R
+SELECT hypot('NaN'::double, 4.0::double);
+----
+NaN
+
+# signed zeros
+query RRR
+SELECT hypot(0.0::double, 0.0::double), hypot(-0.0::double, 0.0::double), hypot(3.0::double, -0.0::double);
+----
+0 0 3
+
+# NULL propagates even when the other input is Infinity
+query R
+SELECT hypot(NULL::double, 'Infinity'::double);
+----
+NULL
+
+# negative inputs yield the positive magnitude
+query RR
+SELECT hypot(-3.0::double, -4.0::double), hypot(-3.0::double, 4.0::double);
+----
+5 5
+
+# Underflow-safe: naive sqrt(a*a + b*b) underflows to 0 for tiny inputs; hypot stays nonzero (matches Spark's Math.hypot)
+query B
+SELECT hypot(3e-200::double, 4e-200::double) > 0;
+----
+true
+
+# Array path with special values (normal, +Infinity, NaN, NULL)
+query R
+SELECT hypot(a, b) FROM (VALUES
+  (3.0::double, 4.0::double),
+  ('Infinity'::double, 1.0::double),
+  ('NaN'::double, 1.0::double),
+  (NULL::double, 1.0::double)) AS t(a, b);
+----
+5
+Infinity
+NaN
+NULL
+
+# both inputs NaN -> NaN
+query R
+SELECT hypot('NaN'::double, 'NaN'::double);
+----
+NaN
+
+# both inputs infinite -> +Infinity
+query R
+SELECT hypot('Infinity'::double, '-Infinity'::double);
+----
+Infinity