diff --git a/e2e_test/v2/batch/func.slt b/e2e_test/v2/batch/func.slt
index 4ddb61177ddf7..fff017dfe95f8 100644
--- a/e2e_test/v2/batch/func.slt
+++ b/e2e_test/v2/batch/func.slt
@@ -69,3 +69,43 @@ drop table t;
 
 statement ok
 drop table b;
+
+query T
+select concat_ws(',', 'a', 'b');
+----
+a,b
+
+query T
+select concat_ws(NULL, 'a', 'b');
+----
+NULL
+
+query T
+select concat_ws(',', NULL, 'b');
+----
+b
+
+query T
+select concat_ws(NULL, NULL, 'b');
+----
+NULL
+
+query T
+select concat_ws(',', 1, 1.01, 'A', true, NULL);
+----
+1,1.01,A,true
+
+statement ok
+create table t (v1 varchar, v2 smallint, v3 int, v4 decimal, v5 real, v6 double, v7 bool, v8 varchar);
+
+statement ok
+insert into t values (',', 1, 2, 3.01, 4, 5.01, true, NULL);
+
+query T
+select concat_ws(v1, v2, v3, v4, v5, v6, v7, v8) from t;
+----
+1,2,3.01,4,5.01,true
+
+
+statement ok
+drop table t;
diff --git a/src/frontend/src/binder/expr/function.rs b/src/frontend/src/binder/expr/function.rs
index acb3a904ba547..354255b9a0a2e 100644
--- a/src/frontend/src/binder/expr/function.rs
+++ b/src/frontend/src/binder/expr/function.rs
@@ -62,6 +62,7 @@ impl Binder {
                     inputs = Self::rewrite_nullif_to_case_when(inputs)?;
                     ExprType::Case
                 }
+                "concat_ws" => ExprType::ConcatWs,
                 "coalesce" => ExprType::Coalesce,
                 "round" => {
                     inputs = Self::rewrite_round_args(inputs);
diff --git a/src/frontend/src/expr/function_call.rs b/src/frontend/src/expr/function_call.rs
index cd11f8340d08b..b6da99bdb760f 100644
--- a/src/frontend/src/expr/function_call.rs
+++ b/src/frontend/src/expr/function_call.rs
@@ -115,6 +115,26 @@ impl FunctionCall {
                 }
                 align_types(inputs.iter_mut())
             }
+            ExprType::ConcatWs => {
+                if inputs.len() < 2 {
+                    return Err(ErrorCode::BindError(
+                        "ConcatWs function must contain at least 2 arguments".into(),
+                    )
+                    .into());
+                }
+
+                inputs = inputs
+                    .into_iter()
+                    .enumerate()
+                    .map(|(i, input)| match i {
+                        // 0-th arg must be string
+                        0 => input.cast_implicit(DataType::Varchar),
+                        // subsequent can be any type
+                        _ => input.cast_explicit(DataType::Varchar),
+                    })
+                    .collect::<Result<Vec<_>>>()?;
+                Ok(DataType::Varchar)
+            }
             _ => infer_type(
                 func_type,
                 inputs.iter().map(|expr| expr.return_type()).collect(),
diff --git a/src/frontend/test_runner/tests/testdata/expr.yaml b/src/frontend/test_runner/tests/testdata/expr.yaml
index 55e39b50e4418..c94887d742af5 100644
--- a/src/frontend/test_runner/tests/testdata/expr.yaml
+++ b/src/frontend/test_runner/tests/testdata/expr.yaml
@@ -215,3 +215,29 @@
     create table t (v1 int);
     select coalesce(1,'a') from t;
   binder_error: 'Bind error: types Int32 and Varchar cannot be matched'
+- sql: |
+    create table t (v1 varchar);
+    select concat_ws(v1, 1) as expr from t;
+  batch_plan: |
+    BatchExchange { order: [], dist: Single }
+      BatchProject { exprs: [ConcatWs($0, 1:Int32::Varchar)] }
+        BatchScan { table: t, columns: [v1] }
+  stream_plan: |
+    StreamMaterialize { columns: [expr, _row_id#0(hidden)], pk_columns: [_row_id#0] }
+      StreamProject { exprs: [ConcatWs($0, 1:Int32::Varchar), $1] }
+        StreamTableScan { table: t, columns: [v1, _row_id#0], pk_indices: [1] }
+- sql: |
+    create table t (v1 varchar);
+    select concat_ws(v1, 1.2) from t;
+  batch_plan: |
+    BatchExchange { order: [], dist: Single }
+      BatchProject { exprs: [ConcatWs($0, 1.2:Decimal::Varchar)] }
+        BatchScan { table: t, columns: [v1] }
+- sql: |
+    create table t (v1 int);
+    select concat_ws(v1, 1.2) from t;
+  binder_error: 'Bind error: cannot cast type Int32 to Varchar in Implicit context'
+- sql: |
+    create table t (v1 int);
+    select concat_ws() from t;
+  binder_error: 'Bind error: ConcatWs function must contain at least 2 arguments'