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",
@@ -61,6 +62,7 @@ ctor = "0.2.0"
 datafusion = { path = "datafusion/core" }
 datafusion-common = { path = "datafusion/common" }
 datafusion-expr = { path = "datafusion/expr" }
+datafusion-functions = { path = "datafusion/functions" }
 datafusion-sql = { path = "datafusion/sql" }
 datafusion-optimizer = { path = "datafusion/optimizer" }
 datafusion-physical-expr = { path = "datafusion/physical-expr" }

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 = "32.0.0", features = ["object_store"], default-features = false }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
+datafusion-functions = { workspace = true }
 datafusion-optimizer = { path = "../optimizer", version = "32.0.0", default-features = false }
 datafusion-physical-expr = { path = "../physical-expr", version = "32.0.0", default-features = false }
 datafusion-physical-plan = { workspace = true }

datafusion/core/src/dataframe/mod.rs

@@ -59,7 +59,7 @@ use crate::logical_expr::{
 use crate::physical_plan::SendableRecordBatchStream;
 use crate::physical_plan::{collect, collect_partitioned};
 use crate::physical_plan::{execute_stream, execute_stream_partitioned, ExecutionPlan};
-use crate::prelude::SessionContext;
+use crate::prelude::*;
 
 /// Contains options that control how data is
 /// written out from a DataFrame

datafusion/core/src/datasource/listing/helpers.rs

@@ -102,7 +102,7 @@ pub fn expr_applicable_for_cols(col_names: &[String], expr: &Expr) -> bool {
                 }
             }
             Expr::ScalarUDF(ScalarUDF { fun, .. }) => {
-                match fun.signature.volatility {
+                match fun.signature().volatility {
                     Volatility::Immutable => VisitRecursion::Continue,
                     // TODO: Stable functions could be `applicable`, but that would require access to the context
                     Volatility::Stable | Volatility::Volatile => {

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

@@ -806,7 +806,7 @@ impl SessionContext {
         self.state
             .write()
             .scalar_functions
-            .insert(f.name.clone(), Arc::new(f));
+            .insert(f.name().to_string(), Arc::new(f));
     }
 
     /// Registers an aggregate UDF within this context.
@@ -1298,14 +1298,18 @@ impl SessionState {
             );
         }
 
+        // register built in functions
+        let mut scalar_functions = HashMap::new();
+        datafusion_functions::register_all(&mut scalar_functions);
+
         SessionState {
             session_id,
             analyzer: Analyzer::new(),
             optimizer: Optimizer::new(),
             physical_optimizers: PhysicalOptimizer::new(),
             query_planner: Arc::new(DefaultQueryPlanner {}),
             catalog_list,
-            scalar_functions: HashMap::new(),
+            scalar_functions,
             aggregate_functions: HashMap::new(),
             window_functions: HashMap::new(),
             serializer_registry: Arc::new(EmptySerializerRegistry),

datafusion/core/src/physical_planner.rs

@@ -222,7 +222,7 @@ fn create_physical_name(e: &Expr, is_first_expr: bool) -> Result<String> {
             create_function_physical_name(&func.fun.to_string(), false, &func.args)
         }
         Expr::ScalarUDF(ScalarUDF { fun, args }) => {
-            create_function_physical_name(&fun.name, false, args)
+            create_function_physical_name(fun.name(), false, args)
         }
         Expr::WindowFunction(WindowFunction { fun, args, .. }) => {
             create_function_physical_name(&fun.to_string(), false, args)

datafusion/expr/src/built_in_function.rs

@@ -72,14 +72,10 @@ pub enum BuiltinScalarFunction {
     Cos,
     /// cos
     Cosh,
-    /// Decode
-    Decode,
     /// degrees
     Degrees,
     /// Digest
     Digest,
-    /// Encode
-    Encode,
     /// exp
     Exp,
     /// factorial
@@ -353,9 +349,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,
@@ -710,30 +704,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")
             }
@@ -1038,24 +1008,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)]),
@@ -1461,10 +1413,6 @@ fn aliases(func: &BuiltinScalarFunction) -> &'static [&'static str] {
         BuiltinScalarFunction::SHA384 => &["sha384"],
         BuiltinScalarFunction::SHA512 => &["sha512"],
 
-        // encode/decode
-        BuiltinScalarFunction::Encode => &["encode"],
-        BuiltinScalarFunction::Decode => &["decode"],
-
         // other functions
         BuiltinScalarFunction::ArrowTypeof => &["arrow_typeof"],