feat(frontend): Bind concat_ws expression

e2e_test/v2/batch/func.slt

@@ -69,3 +69,33 @@ drop table t;
 
 statement ok
 drop table b;
+
+query T
+select concat_ws(',', 'a', 'b');
+----
+a,b
+
+query T
+select concat_ws(',', NULL, 'b');
+----
+b
+
+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;

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);

src/frontend/src/expr/function_call.rs

@@ -115,6 +115,30 @@ 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());
+                }
+
+                if inputs[0].return_type() != DataType::Varchar {
+                    return Err(ErrorCode::BindError(
+                        "ConcatWs function must have text as first argument".into(),
+                    )
+                    .into());
+                }
+
+                // subsequent inputs can be any type, they are cast into varchars with
+                // explicit_cast.
+                inputs = inputs
+                    .into_iter()
+                    .map(|input| input.cast_explicit(DataType::Varchar))
+                    .collect::<Result<Vec<_>>>()?;
+
+                Ok(DataType::Varchar)
+            }
             _ => infer_type(
                 func_type,
                 inputs.iter().map(|expr| expr.return_type()).collect(),

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: ConcatWs function must have text as first argument'
+- sql: |
+    create table t (v1 int);
+    select concat_ws() from t;
+  binder_error: 'Bind error: ConcatWs function must contain at least 2 arguments'