feat(expr): implementation for char_length operation

e2e_test/batch/func.slt

@@ -135,3 +135,13 @@ select split_part('abc,def,ghi,jkl', ',', -2);
 ----
 ghi
 
+
+query I
+select char_length('hello world');
+----
+11
+
+query I
+select char_length('abcdefghijklmnopqrstuvwxyz');
+----
+26

proto/expr.proto

@@ -66,8 +66,8 @@ message ExprNode {
     CEIL = 221;
     FLOOR = 222;
     TO_CHAR = 223;
-
     MD5 = 224;
+    CHAR_LENGTH = 225;
 
     // Boolean comparison
     IS_TRUE = 301;

src/expr/src/expr/expr_unary.rs

@@ -289,6 +289,11 @@ pub fn new_unary_expr(
             return_type,
             ascii,
         )),
+        (ProstType::CharLength, _, _) => Box::new(UnaryExpression::<Utf8Array, I32Array, _>::new(
+            child_expr,
+            return_type,
+            length_default,
+        )),
         (ProstType::Neg, _, _) => {
             gen_unary_atm_expr! { "Neg", child_expr, return_type, general_neg,
                 {
@@ -334,7 +339,7 @@ pub fn new_unary_expr(
 }
 
 pub fn new_length_default(expr_ia1: BoxedExpression, return_type: DataType) -> BoxedExpression {
-    Box::new(UnaryExpression::<Utf8Array, I64Array, _>::new(
+    Box::new(UnaryExpression::<Utf8Array, I32Array, _>::new(
         expr_ia1,
         return_type,
         length_default,

src/expr/src/expr/mod.rs

@@ -80,7 +80,7 @@ pub fn build_from_prost(prost: &ExprNode) -> Result<BoxedExpression> {
 
     match prost.get_expr_type()? {
         Cast | Upper | Lower | Md5 | Not | IsTrue | IsNotTrue | IsFalse | IsNotFalse | IsNull
-        | IsNotNull | Neg | Ascii | Abs | Ceil | Floor | Round | BitwiseNot => {
+        | IsNotNull | Neg | Ascii | Abs | Ceil | Floor | Round | BitwiseNot | CharLength => {
             build_unary_expr_prost(prost)
         }
         Equal | NotEqual | LessThan | LessThanOrEqual | GreaterThan | GreaterThanOrEqual | Add

src/expr/src/vector_op/length.rs

@@ -15,8 +15,8 @@
 use risingwave_common::error::Result;
 
 #[inline(always)]
-pub fn length_default(s: &str) -> Result<i64> {
-    Ok(s.chars().count() as i64)
+pub fn length_default(s: &str) -> Result<i32> {
+    Ok(s.chars().count() as i32)
 }
 
 #[cfg(test)]

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

@@ -95,6 +95,8 @@ impl Binder {
                     inputs = Self::rewrite_two_bool_inputs(inputs)?;
                     ExprType::NotEqual
                 }
+                "char_length" => ExprType::CharLength,
+                "character_length" => ExprType::CharLength,
                 _ => {
                     return Err(ErrorCode::NotImplemented(
                         format!("unsupported function: {:?}", function_name),

src/frontend/src/expr/type_inference.rs

@@ -341,7 +341,7 @@ fn build_type_derive_map() -> HashMap<FuncSign, DataTypeName> {
             T::Varchar,
         );
     }
-    for e in [E::Length, E::Ascii] {
+    for e in [E::Length, E::Ascii, E::CharLength] {
         map.insert(FuncSign::new(e, vec![T::Varchar]), T::Int32);
     }
     map.insert(