RFC: Demonstrate what a function package might look like -- encoding expressions

Cargo.toml

@@ -22,6 +22,7 @@ members = [
     "datafusion/core",
     "datafusion/expr",
     "datafusion/execution",
+    "datafusion/functions",
     "datafusion/optimizer",
     "datafusion/physical-expr",
     "datafusion/physical-plan",
@@ -62,6 +63,7 @@ ctor = "0.2.0"
 datafusion = { path = "datafusion/core", version = "34.0.0" }
 datafusion-common = { path = "datafusion/common", version = "34.0.0" }
 datafusion-expr = { path = "datafusion/expr", version = "34.0.0" }
+datafusion-functions = { path = "datafusion/functions", version = "34.0.0" }
 datafusion-sql = { path = "datafusion/sql", version = "34.0.0" }
 datafusion-optimizer = { path = "datafusion/optimizer", version = "34.0.0" }
 datafusion-physical-expr = { path = "datafusion/physical-expr", version = "34.0.0" }

datafusion/core/Cargo.toml

@@ -40,7 +40,7 @@ backtrace = ["datafusion-common/backtrace"]
 compression = ["xz2", "bzip2", "flate2", "zstd", "async-compression"]
 crypto_expressions = ["datafusion-physical-expr/crypto_expressions", "datafusion-optimizer/crypto_expressions"]
 default = ["crypto_expressions", "encoding_expressions", "regex_expressions", "unicode_expressions", "compression", "parquet"]
-encoding_expressions = ["datafusion-physical-expr/encoding_expressions"]
+encoding_expressions = ["datafusion-functions/encoding_expressions"]
 # Used for testing ONLY: causes all values to hash to the same value (test for collisions)
 force_hash_collisions = []
 parquet = ["datafusion-common/parquet", "dep:parquet"]
@@ -65,6 +65,7 @@ dashmap = { workspace = true }
 datafusion-common = { path = "../common", version = "34.0.0", features = ["object_store"], default-features = false }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
+datafusion-functions = { path = "../functions", version = "34.0.0"}
 datafusion-optimizer = { path = "../optimizer", version = "34.0.0", default-features = false }
 datafusion-physical-expr = { path = "../physical-expr", version = "34.0.0", default-features = false }
 datafusion-physical-plan = { workspace = true }

datafusion/core/src/dataframe/mod.rs

@@ -40,7 +40,6 @@ use crate::physical_plan::{
     collect, collect_partitioned, execute_stream, execute_stream_partitioned,
     ExecutionPlan, SendableRecordBatchStream,
 };
-use crate::prelude::SessionContext;
 
 use arrow::array::{Array, ArrayRef, Int64Array, StringArray};
 use arrow::compute::{cast, concat};
@@ -59,6 +58,7 @@ use datafusion_expr::{
     TableProviderFilterPushDown, UNNAMED_TABLE,
 };
 
+use crate::prelude::SessionContext;
 use async_trait::async_trait;
 
 /// Contains options that control how data is

datafusion/core/src/execution/context/mod.rs

@@ -1343,7 +1343,7 @@ impl SessionState {
             );
         }
 
-        SessionState {
+        let mut new_self = SessionState {
             session_id,
             analyzer: Analyzer::new(),
             optimizer: Optimizer::new(),
@@ -1359,7 +1359,13 @@ impl SessionState {
             execution_props: ExecutionProps::new(),
             runtime_env: runtime,
             table_factories,
-        }
+        };
+
+        // register built in functions
+        datafusion_functions::register_all(&mut new_self)
+            .expect("can not register built in functions");
+
+        new_self
     }
     /// Returns new [`SessionState`] using the provided
     /// [`SessionConfig`] and [`RuntimeEnv`].
@@ -1968,6 +1974,10 @@ impl FunctionRegistry for SessionState {
             plan_datafusion_err!("There is no UDWF named \"{name}\" in the registry")
         })
     }
+
+    fn register_udf(&mut self, udf: Arc<ScalarUDF>) -> Result<Option<Arc<ScalarUDF>>> {
+        Ok(self.scalar_functions.insert(udf.name().into(), udf))
+    }
 }
 
 impl OptimizerConfig for SessionState {

datafusion/core/src/prelude.rs

@@ -38,6 +38,7 @@ pub use datafusion_expr::{
     logical_plan::{JoinType, Partitioning},
     Expr,
 };
+pub use datafusion_functions::expr_fn::*;
 
 pub use std::ops::Not;
 pub use std::ops::{Add, Div, Mul, Neg, Rem, Sub};

datafusion/execution/src/registry.rs

@@ -17,7 +17,7 @@
 
 //! FunctionRegistry trait
 
-use datafusion_common::Result;
+use datafusion_common::{not_impl_err, DataFusionError, Result};
 use datafusion_expr::{AggregateUDF, ScalarUDF, UserDefinedLogicalNode, WindowUDF};
 use std::{collections::HashSet, sync::Arc};
 
@@ -34,6 +34,17 @@ pub trait FunctionRegistry {
 
     /// Returns a reference to the udwf named `name`.
     fn udwf(&self, name: &str) -> Result<Arc<WindowUDF>>;
+
+    /// Registers a new `ScalarUDF`, returning any previously registered
+    /// implementation.
+    ///
+    /// Returns an error (default) if the function can not be registered, for
+    /// example because the registry doesn't support new functions
+    fn register_udf(&mut self, _udf: Arc<ScalarUDF>) -> Result<Option<Arc<ScalarUDF>>> {
+        not_impl_err!("Registering ScalarUDF")
+    }
+
+    // TODO add register_udaf and register_udwf
 }
 
 /// Serializer and deserializer registry for extensions like [UserDefinedLogicalNode].

datafusion/expr/src/built_in_function.rs

@@ -69,14 +69,10 @@ pub enum BuiltinScalarFunction {
     Cos,
     /// cos
     Cosh,
-    /// Decode
-    Decode,
     /// degrees
     Degrees,
     /// Digest
     Digest,
-    /// Encode
-    Encode,
     /// exp
     Exp,
     /// factorial
@@ -373,9 +369,7 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::Coalesce => Volatility::Immutable,
             BuiltinScalarFunction::Cos => Volatility::Immutable,
             BuiltinScalarFunction::Cosh => Volatility::Immutable,
-            BuiltinScalarFunction::Decode => Volatility::Immutable,
             BuiltinScalarFunction::Degrees => Volatility::Immutable,
-            BuiltinScalarFunction::Encode => Volatility::Immutable,
             BuiltinScalarFunction::Exp => Volatility::Immutable,
             BuiltinScalarFunction::Factorial => Volatility::Immutable,
             BuiltinScalarFunction::Floor => Volatility::Immutable,
@@ -746,30 +740,6 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::Digest => {
                 utf8_or_binary_to_binary_type(&input_expr_types[0], "digest")
             }
-            BuiltinScalarFunction::Encode => Ok(match input_expr_types[0] {
-                Utf8 => Utf8,
-                LargeUtf8 => LargeUtf8,
-                Binary => Utf8,
-                LargeBinary => LargeUtf8,
-                Null => Null,
-                _ => {
-                    return plan_err!(
-                        "The encode function can only accept utf8 or binary."
-                    );
-                }
-            }),
-            BuiltinScalarFunction::Decode => Ok(match input_expr_types[0] {
-                Utf8 => Binary,
-                LargeUtf8 => LargeBinary,
-                Binary => Binary,
-                LargeBinary => LargeBinary,
-                Null => Null,
-                _ => {
-                    return plan_err!(
-                        "The decode function can only accept utf8 or binary."
-                    );
-                }
-            }),
             BuiltinScalarFunction::SplitPart => {
                 utf8_to_str_type(&input_expr_types[0], "split_part")
             }
@@ -1100,24 +1070,6 @@ impl BuiltinScalarFunction {
                 ],
                 self.volatility(),
             ),
-            BuiltinScalarFunction::Encode => Signature::one_of(
-                vec![
-                    Exact(vec![Utf8, Utf8]),
-                    Exact(vec![LargeUtf8, Utf8]),
-                    Exact(vec![Binary, Utf8]),
-                    Exact(vec![LargeBinary, Utf8]),
-                ],
-                self.volatility(),
-            ),
-            BuiltinScalarFunction::Decode => Signature::one_of(
-                vec![
-                    Exact(vec![Utf8, Utf8]),
-                    Exact(vec![LargeUtf8, Utf8]),
-                    Exact(vec![Binary, Utf8]),
-                    Exact(vec![LargeBinary, Utf8]),
-                ],
-                self.volatility(),
-            ),
             BuiltinScalarFunction::DateTrunc => Signature::one_of(
                 vec![
                     Exact(vec![Utf8, Timestamp(Nanosecond, None)]),
@@ -1552,10 +1504,6 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::SHA384 => &["sha384"],
             BuiltinScalarFunction::SHA512 => &["sha512"],
 
-            // encode/decode
-            BuiltinScalarFunction::Encode => &["encode"],
-            BuiltinScalarFunction::Decode => &["decode"],
-
             // other functions
             BuiltinScalarFunction::ArrowTypeof => &["arrow_typeof"],