feat: expose Spark-compatible functions (apache#1482)

Add `datafusion.functions.spark` module exposing the upstream
`datafusion-spark` crate's UDF/UDAF library (~87 functions across string,
math, datetime, hash, array, aggregate, bitwise, bitmap, conditional,
collection, conversion, json, map, url categories).

For DataFrame use, import the typed Python wrappers from
`datafusion.functions.spark`. For SQL use, call
`SessionContext.enable_spark_functions()` to register the Spark UDFs by
name (overriding DataFusion built-ins of the same name with their Spark
semantics — NULL-propagating `concat`, 1-indexed `substring`, HALF_UP
`round`, etc.).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: timsaucer

Cargo.lock

@@ -48,6 +48,7 @@ datafusion-catalog = { version = "54", default-features = false }
 datafusion-common = { version = "54", default-features = false }
 datafusion-functions-aggregate = { version = "54" }
 datafusion-functions-window = { version = "54" }
+datafusion-spark = { version = "54" }
 datafusion-expr = { version = "54" }
 prost = "0.14.3"
 serde_json = "1"
@@ -79,4 +80,5 @@ datafusion-catalog = { git = "https://github.com/apache/datafusion", rev = "1321
 datafusion-common = { git = "https://github.com/apache/datafusion", rev = "1321d60cc37ee487d1e7ce7f501357c3236b2542" }
 datafusion-functions-aggregate = { git = "https://github.com/apache/datafusion", rev = "1321d60cc37ee487d1e7ce7f501357c3236b2542" }
 datafusion-functions-window = { git = "https://github.com/apache/datafusion", rev = "1321d60cc37ee487d1e7ce7f501357c3236b2542" }
+datafusion-spark = { git = "https://github.com/apache/datafusion", rev = "1321d60cc37ee487d1e7ce7f501357c3236b2542" }
 datafusion-expr = { git = "https://github.com/apache/datafusion", rev = "1321d60cc37ee487d1e7ce7f501357c3236b2542" }

Cargo.toml

@@ -53,6 +53,7 @@ datafusion = { workspace = true, features = ["avro", "unicode_expressions"] }
 datafusion-substrait = { workspace = true, optional = true }
 datafusion-proto = { workspace = true }
 datafusion-ffi = { workspace = true }
+datafusion-spark = { workspace = true }
 prost = { workspace = true } # keep in line with `datafusion-substrait`
 serde_json = { workspace = true }
 uuid = { workspace = true, features = ["v4"] }

crates/core/Cargo.toml

@@ -1059,6 +1059,21 @@ impl PySessionContext {
         Ok(())
     }
 
+    /// Register all `datafusion-spark` UDFs/UDAFs/UDWFs, overriding any built-in
+    /// DataFusion functions of the same name with their Spark-semantics version.
+    pub fn enable_spark_functions(&self) -> PyResult<()> {
+        for udf in datafusion_spark::all_default_scalar_functions() {
+            self.ctx.register_udf((*udf).clone());
+        }
+        for udaf in datafusion_spark::all_default_aggregate_functions() {
+            self.ctx.register_udaf((*udaf).clone());
+        }
+        for udwf in datafusion_spark::all_default_window_functions() {
+            self.ctx.register_udwf((*udwf).clone());
+        }
+        Ok(())
+    }
+
     pub fn deregister_udaf(&self, name: &str) {
         self.ctx.deregister_udaf(name);
     }

crates/core/src/context.rs

@@ -31,7 +31,7 @@ use crate::expr::conditional_expr::PyCaseBuilder;
 use crate::expr::sort_expr::{PySortExpr, to_sort_expressions};
 use crate::expr::window::PyWindowFrame;
 
-fn add_builder_fns_to_aggregate(
+pub(crate) fn add_builder_fns_to_aggregate(
     agg_fn: Expr,
     distinct: Option<bool>,
     filter: Option<PyExpr>,

crates/core/src/functions.rs

@@ -48,6 +48,8 @@ pub mod physical_plan;
 mod pyarrow_filter_expression;
 pub mod pyarrow_util;
 mod record_batch;
+#[allow(clippy::borrow_deref_ref)]
+mod spark_functions;
 pub mod sql;
 pub mod store;
 pub mod table;
@@ -123,6 +125,10 @@ fn _internal(py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
     // Register the functions as a submodule
     let funcs = PyModule::new(py, "functions")?;
     functions::init_module(&funcs)?;
+    // Spark-compatible functions live under `functions.spark`.
+    let spark_funcs = PyModule::new(py, "spark")?;
+    spark_functions::init_module(&spark_funcs)?;
+    funcs.add_submodule(&spark_funcs)?;
     m.add_submodule(&funcs)?;
 
     let store = PyModule::new(py, "object_store")?;