feat: add Spark-compatible hypot function

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.
+///
+/// <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; `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<DataType> {
+        Ok(DataType::Float64)
+    }
+
+    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+        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::<Float64Type>();
+        let y = y.as_primitive::<Float64Type>();
+
+        // `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))
+    }
+}

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,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