feat(sql-udf): support basic anonymous sql udf

e2e_test/udf/sql_udf.slt

@@ -0,0 +1,105 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+# Create an anonymous function with double dollar as clause
+statement ok
+create function add(INT, INT) returns int as $$select $1 + $2$$ language sql;
+
+# Create an anonymous function with single quote as clause
+statement ok
+create function sub(INT, INT) returns int as 'select $1 - $2' language sql;
+
+# Currently we can only support constant calling convention
+statement ok
+create function add_sub_binding() returns int as 'select add(1, 1) + sub(2, 2)' language sql;
+
+# Call the defined sql udf
+query I
+select add(1, -1);
+----
+0
+
+query I
+select sub(1, 1);
+----
+0
+
+query I
+select add_sub_binding();
+----
+2
+
+query III
+select add(1, -1), sub(1, 1), add_sub_binding();
+----
+0 0 2
+
+# Create a mock table
+statement ok
+create table t1 (c1 INT, c2 INT);
+
+# Insert some data into the mock table
+statement ok
+insert into t1 values (1, 1), (2, 2), (3, 3), (4, 4), (5, 5);
+
+query III
+select sub(c1, c2), c1, c2, add(c1, c2) from t1 order by c1 asc;
+----
+0 1 1 2
+0 2 2 4
+0 3 3 6
+0 4 4 8
+0 5 5 10
+
+# Invalid function body syntax
+statement error
+create function add_error(INT, INT) returns int as $$select $1 + $2 +$$ language sql;
+
+# Multiple type interleaving sql udf
+statement ok
+create function add_sub(INT, FLOAT, INT) returns FLOAT as $$select -$1 + $2 - $3$$ language sql;
+
+# Note: need EXPLICIT type cast in order to call the multiple types interleaving sql udf
+query I
+select add_sub(1::INT, 5.1415926::FLOAT, 1::INT);
+----
+3.1415926
+
+query III
+select add(1, -1), sub(1, 1), add_sub(1::INT, 5.1415926::FLOAT, 1::INT);
+----
+0 0 3.1415926
+
+# Create another mock table
+statement ok
+create table t2 (c1 INT, c2 FLOAT, c3 INT);
+
+statement ok
+insert into t2 values (1, 3.14, 2), (2, 4.44, 5), (20, 10.30, 02);
+
+query IIIII
+select c1, c2, c3, add(c1, c3), sub(c1, c3), add_sub(c1::INT, c2::FLOAT, c3::INT) from t2 order by c1 asc;
+----
+1 3.14 2 3 -1 0.14000000000000012
+2 4.44 5 7 -3 -2.5599999999999996
+20 10.3 2 22 18 -11.7
+
+# Drop the functions
+statement ok
+drop function add;
+
+statement ok
+drop function sub;
+
+statement ok
+drop function add_sub_binding;
+
+statement ok
+drop function add_sub;
+
+# Drop the mock table
+statement ok
+drop table t1;
+
+statement ok
+drop table t2;

src/frontend/src/binder/expr/column.rs

@@ -37,6 +37,11 @@ impl Binder {
             }
         };
 
+        // Special check for sql udf
+        if self.udf_context.contains_key(&column_name) {
+            return self.bind_expr(self.udf_context.get(&column_name).unwrap().clone());
+        }
+
         match self
             .context
             .get_column_binding_indices(&table_name, &column_name)

src/frontend/src/binder/expr/function.rs

@@ -15,7 +15,7 @@
 use std::collections::HashMap;
 use std::iter::once;
 use std::str::FromStr;
-use std::sync::LazyLock;
+use std::sync::{Arc, LazyLock};
 
 use bk_tree::{metrics, BKTree};
 use itertools::Itertools;
@@ -30,13 +30,14 @@ use risingwave_expr::window_function::{
     Frame, FrameBound, FrameBounds, FrameExclusion, WindowFuncKind,
 };
 use risingwave_sqlparser::ast::{
-    self, Function, FunctionArg, FunctionArgExpr, Ident, WindowFrameBound, WindowFrameExclusion,
-    WindowFrameUnits, WindowSpec,
+    self, Expr as AstExpr, Function, FunctionArg, FunctionArgExpr, Ident, SelectItem, SetExpr,
+    Statement, WindowFrameBound, WindowFrameExclusion, WindowFrameUnits, WindowSpec,
 };
 use thiserror_ext::AsReport;
 
 use crate::binder::bind_context::Clause;
 use crate::binder::{Binder, BoundQuery, BoundSetExpr};
+use crate::catalog::function_catalog::FunctionCatalog;
 use crate::expr::{
     AggCall, Expr, ExprImpl, ExprType, FunctionCall, FunctionCallWithLambda, Literal, Now, OrderBy,
     Subquery, SubqueryKind, TableFunction, TableFunctionType, UserDefinedFunction, WindowFunction,
@@ -117,6 +118,9 @@ impl Binder {
             return self.bind_array_transform(f);
         }
 
+        // Used later in sql udf expression evaluation
+        let args = f.args.clone();
+
         let inputs = f
             .args
             .into_iter()
@@ -149,6 +153,53 @@ impl Binder {
             return Ok(TableFunction::new(function_type, inputs)?.into());
         }
 
+        /// TODO: add name related logic
+        fn create_udf_context(
+            binder: &mut Binder,
+            args: &[FunctionArg],
+            _catalog: &Arc<FunctionCatalog>,
+        ) {
+            binder.udf_context = args
+                .iter()
+                .enumerate()
+                .map(|(i, current_arg)| {
+                    if let FunctionArg::Unnamed(arg) = current_arg {
+                        let FunctionArgExpr::Expr(e) = arg else {
+                            panic!("invalid syntax");
+                        };
+                        // if catalog.arg_names.is_some() {
+                        //     panic!("invalid syntax");
+                        // }
+                        return ("$".to_string() + &(i + 1).to_string(), e.clone());
+                    }
+                    panic!("invalid syntax");
+                })
+                .collect()
+        }
+
+        fn extract_udf_expression(ast: Vec<Statement>) -> AstExpr {
+            // Extract the expression out
+            let Statement::Query(query) = ast
+                .into_iter()
+                .exactly_one()
+                .expect("sql udf should contain only one statement")
+            else {
+                unreachable!("sql udf should contain a query statement");
+            };
+            let SetExpr::Select(select) = query.body else {
+                panic!("Invalid syntax");
+            };
+            let projection = select.projection;
+            let SelectItem::UnnamedExpr(expr) = projection
+                .into_iter()
+                .exactly_one()
+                .expect("`projection` should contain only one `SelectItem`")
+            else {
+                unreachable!("`projection` should contain only one `SelectItem`");
+            };
+            expr
+        }
+
         // user defined function
         // TODO: resolve schema name https://github.com/risingwavelabs/risingwave/issues/12422
         if let Ok(schema) = self.first_valid_schema()
@@ -158,13 +209,22 @@ impl Binder {
             )
         {
             use crate::catalog::function_catalog::FunctionKind::*;
-            match &func.kind {
-                Scalar { .. } => return Ok(UserDefinedFunction::new(func.clone(), inputs).into()),
-                Table { .. } => {
-                    self.ensure_table_function_allowed()?;
-                    return Ok(TableFunction::new_user_defined(func.clone(), inputs).into());
+            if func.language == "sql" {
+                // This represents the current user defined function is `language sql`
+                let ast = risingwave_sqlparser::parser::Parser::parse_sql(func.identifier.as_str()).unwrap();
+                // The actual inline logic
+                create_udf_context(self, &args, &Arc::clone(func));
+                return self.bind_expr(extract_udf_expression(ast));
+            } else {
+                debug_assert!(func.language == "python" || func.language == "java", "only `python` and `java` are currently supported for general udf");
+                match &func.kind {
+                    Scalar { .. } => return Ok(UserDefinedFunction::new(func.clone(), inputs).into()),
+                    Table { .. } => {
+                        self.ensure_table_function_allowed()?;
+                        return Ok(TableFunction::new_user_defined(func.clone(), inputs).into());
+                    }
+                    Aggregate => todo!("support UDAF"),
                 }
-                Aggregate => todo!("support UDAF"),
             }
         }
 
@@ -1216,7 +1276,7 @@ impl Binder {
         static FUNCTIONS_BKTREE: LazyLock<BKTree<&str>> = LazyLock::new(|| {
             let mut tree = BKTree::new(metrics::Levenshtein);
 
-            // TODO: Also hint other functinos, e,g, Agg or UDF.
+            // TODO: Also hint other functinos, e.g., Agg or UDF.
             for k in HANDLES.keys() {
                 tree.add(*k);
             }

src/frontend/src/binder/expr/mod.rs

@@ -378,6 +378,12 @@ impl Binder {
     }
 
     fn bind_parameter(&mut self, index: u64) -> Result<ExprImpl> {
+        // Special check for sql udf
+        let name = "$".to_string() + &index.to_string();
+        if self.udf_context.contains_key(&name) {
+            return self.bind_expr(self.udf_context.get(&name).unwrap().clone());
+        }
+
         Ok(Parameter::new(index, self.param_types.clone()).into())
     }
 

src/frontend/src/binder/mod.rs

@@ -21,7 +21,7 @@ use risingwave_common::error::Result;
 use risingwave_common::session_config::{ConfigMap, SearchPath};
 use risingwave_common::types::DataType;
 use risingwave_common::util::iter_util::ZipEqDebug;
-use risingwave_sqlparser::ast::Statement;
+use risingwave_sqlparser::ast::{Expr as AstExpr, Statement};
 
 mod bind_context;
 mod bind_param;
@@ -115,6 +115,8 @@ pub struct Binder {
     included_relations: HashSet<TableId>,
 
     param_types: ParameterTypes,
+
+    udf_context: HashMap<String, AstExpr>,
 }
 
 /// `ParameterTypes` is used to record the types of the parameters during binding. It works
@@ -216,6 +218,7 @@ impl Binder {
             shared_views: HashMap::new(),
             included_relations: HashSet::new(),
             param_types: ParameterTypes::new(param_types),
+            udf_context: HashMap::new(),
         }
     }
 

src/frontend/src/catalog/root_catalog.rs

@@ -88,7 +88,7 @@ impl<'a> SchemaPath<'a> {
 /// - catalog (root catalog)
 ///   - database catalog
 ///     - schema catalog
-///       - function catalog
+///       - function catalog (e.g., user defined function)
 ///       - table/sink/source/index/view catalog
 ///        - column catalog
 pub struct Catalog {

src/frontend/src/catalog/schema_catalog.rs

@@ -542,6 +542,7 @@ impl SchemaCatalog {
         name: &str,
         args: &[DataType],
     ) -> Option<&Arc<FunctionCatalog>> {
+        println!("Current args: {:#?}", args);
         self.function_by_name.get(name)?.get(args)
     }
 

src/frontend/src/handler/create_function.rs

@@ -44,6 +44,7 @@ pub async fn handle_create_function(
     if temporary {
         bail_not_implemented!("CREATE TEMPORARY FUNCTION");
     }
+    // e.g., `language [ python / java / ...etc]`
     let language = match params.language {
         Some(lang) => {
             let lang = lang.real_value().to_lowercase();