diff --git a/be/src/exec/data_sink.cpp b/be/src/exec/data_sink.cpp
index 4fd6bf3b793fa3..c10bb0c18912a7 100644
--- a/be/src/exec/data_sink.cpp
+++ b/be/src/exec/data_sink.cpp
@@ -80,7 +80,7 @@ Status DataSink::create_data_sink(ObjectPool* pool, const TDataSink& thrift_sink
         if (is_vec) {
             tmp_sink = new doris::vectorized::ResultSink(row_desc, output_exprs, thrift_sink.result_sink, 1024);
         } else {
-            tmp_sink = new ResultSink(row_desc, output_exprs, thrift_sink.result_sink, 1024, config::is_vec);
+            tmp_sink = new ResultSink(row_desc, output_exprs, thrift_sink.result_sink, 1024);
         }
         sink->reset(tmp_sink);
         break;
diff --git a/be/src/runtime/mysql_result_writer.cpp b/be/src/runtime/mysql_result_writer.cpp
index f906b71c9a6a8f..80916f5a7faf6a 100644
--- a/be/src/runtime/mysql_result_writer.cpp
+++ b/be/src/runtime/mysql_result_writer.cpp
@@ -39,23 +39,15 @@
 namespace doris {
 
 MysqlResultWriter::MysqlResultWriter(BufferControlBlock* sinker,
-                                     const std::vector<ExprContext*>& output_expr_ctxs, const std::vector<vectorized::VExprContext*>& output_vexpr_ctxs,
-                                     RuntimeProfile* parent_profile)
-        : ResultWriter(!output_vexpr_ctxs.empty()),
+                                     const std::vector<ExprContext*>& output_expr_ctxs, RuntimeProfile* parent_profile)
+        : ResultWriter(),
           _sinker(sinker),
           _output_expr_ctxs(output_expr_ctxs),
-          _output_vexpr_ctxs(output_vexpr_ctxs),
           _row_buffer(NULL),
           _parent_profile(parent_profile) {}
 
 MysqlResultWriter::~MysqlResultWriter() {
-    if (_is_vec) {
-        for (auto buffer : _vec_buffers) {
-            delete buffer;
-        }
-    } else {
-        delete _row_buffer;
-    }
+    delete _row_buffer;
 }
 
 Status MysqlResultWriter::init(RuntimeState* state) {
@@ -64,16 +56,9 @@ Status MysqlResultWriter::init(RuntimeState* state) {
         return Status::InternalError("sinker is NULL pointer.");
     }
 
-    if (_is_vec) {
-        _vec_buffers.resize(state->batch_size());
-        for (int i = 0; i < state->batch_size() ; ++i) {
-            _vec_buffers[i] = new MysqlRowBuffer();
-        }
-    } else {
-        _row_buffer = new(std::nothrow) MysqlRowBuffer();
-        if (NULL == _row_buffer) {
-            return Status::InternalError("no memory to alloc.");
-        }
+    _row_buffer = new(std::nothrow) MysqlRowBuffer();
+    if (NULL == _row_buffer) {
+        return Status::InternalError("no memory to alloc.");
     }
 
     return Status::OK();
@@ -246,118 +231,7 @@ Status MysqlResultWriter::_add_one_row(TupleRow* row) {
     return Status::OK();
 }
 
-template <PrimitiveType type, bool is_nullable>
-Status MysqlResultWriter::_add_one_column(const doris::vectorized::ColumnPtr& column_ptr) {
-    SCOPED_TIMER(_convert_tuple_timer);
-    for (const auto buffer : _vec_buffers) {
-        buffer->reset();
-    }
-
-    doris::vectorized::ColumnPtr column;
-    if constexpr (is_nullable) {
-        column = assert_cast<const vectorized::ColumnNullable&>(*column_ptr).getNestedColumnPtr();
-    } else {
-        column = column_ptr;
-    }
-
-    int buf_ret = 0;
-    for (int i = 0; i < column_ptr->size(); ++i) {
-        if constexpr (is_nullable) {
-            if (column_ptr->isNullAt(i)) {
-                buf_ret = _vec_buffers[i]->push_null();
-                continue;
-            }
-        }
-
-        if constexpr (type == TYPE_TINYINT) {
-            buf_ret = _vec_buffers[i]->push_tinyint(assert_cast<const vectorized::ColumnVector<vectorized::Int8>&>(*column).getData()[i]);
-        }
-        if constexpr (type == TYPE_SMALLINT) {
-            buf_ret = _vec_buffers[i]->push_smallint(assert_cast<const vectorized::ColumnVector<vectorized::Int16>&>(*column).getData()[i]);
-        }
-        if constexpr (type == TYPE_INT) {
-            buf_ret = _vec_buffers[i]->push_int(assert_cast<const vectorized::ColumnVector<vectorized::Int32>&>(*column).getData()[i]);
-        }
-        if constexpr (type == TYPE_BIGINT) {
-            buf_ret = _vec_buffers[i]->push_bigint(assert_cast<const vectorized::ColumnVector<vectorized::Int64>&>(*column).getData()[i]);
-        }
-        if constexpr (type == TYPE_LARGEINT) {
-            char buf[48];
-            int len = 48;
-            char* v = LargeIntValue::to_string(assert_cast<const vectorized::ColumnVector<vectorized::Int128>&>(*column).getData()[i],
-                                               buf, &len);
-            buf_ret = _vec_buffers[i]->push_string(v, len);
-        }
-        if constexpr (type == TYPE_FLOAT) {
-            buf_ret = _vec_buffers[i]->push_float(assert_cast<const vectorized::ColumnVector<vectorized::Float32>&>(*column).getData()[i]);
-        }
-        if constexpr (type == TYPE_DOUBLE) {
-            buf_ret = _vec_buffers[i]->push_double(assert_cast<const vectorized::ColumnVector<vectorized::Float64>&>(*column).getData()[i]);
-        }
-        if constexpr (type == TYPE_DATETIME) {
-            char buf[64];
-            auto time_num = assert_cast<const vectorized::ColumnVector<vectorized::Int128>&>(*column).getData()[i];
-            DateTimeValue time_val;
-            memcpy(&time_val, &time_num, sizeof(vectorized::Int128));
-            // TODO(zhaochun), this function has core risk
-            char* pos = time_val.to_string(buf);
-            buf_ret = _vec_buffers[i]->push_string(buf, pos - buf - 1);
-        }
-
-        if constexpr (type == TYPE_OBJECT) {
-            buf_ret = _vec_buffers[i]->push_null();
-        }
-        if constexpr (type == TYPE_VARCHAR) {
-            const auto string_val = column->getDataAt(i);
-
-            if (string_val.data == NULL) {
-                if (string_val.size == 0) {
-                    // 0x01 is a magic num, not useful actually, just for present ""
-                    char* tmp_val = reinterpret_cast<char*>(0x01);
-                    buf_ret = _vec_buffers[i]->push_string(tmp_val, string_val.size);
-                } else {
-                    buf_ret = _vec_buffers[i]->push_null();
-                }
-            } else {
-                buf_ret = _vec_buffers[i]->push_string(string_val.data, string_val.size);
-            }
-        }
-        if constexpr (type ==  TYPE_DECIMALV2) {
-            DecimalV2Value decimal_val(assert_cast<const vectorized::ColumnDecimal<vectorized::Decimal128>&>(*column).getData()[i]);
-            std::string decimal_str;
-//            int output_scale = _output_expr_ctxs[i]->root()->output_scale();
-//
-//            if (output_scale > 0 && output_scale <= 30) {
-//                decimal_str = decimal_val.to_string(output_scale);
-//            } else {
-                decimal_str = decimal_val.to_string();
-//            }
-            buf_ret = _vec_buffers[i]->push_string(decimal_str.c_str(), decimal_str.length());
-        }
-
-        if (0 != buf_ret) {
-            return Status::InternalError("pack mysql buffer failed.");
-        }
-    }
-
-    return Status::OK();
-}
-
 Status MysqlResultWriter::append_row_batch(const RowBatch* batch) {
-    if (_is_vec) {
-        auto block = batch->convert_to_vec_block();
-
-        _result_column_ids.resize(_output_vexpr_ctxs.size());
-        for (int i = 0; i < _output_vexpr_ctxs.size(); i++) {
-            const auto& vexpr_ctx = _output_vexpr_ctxs[i];
-            int result_column_id = -1;
-            vexpr_ctx->execute(&block, &result_column_id);
-            DCHECK(result_column_id != -1);
-            _result_column_ids[i] = result_column_id;
-        }
-        return append_block(block);
-    }
-
     SCOPED_TIMER(_append_row_batch_timer);
     if (NULL == batch || 0 == batch->num_rows()) {
         return Status::OK();
@@ -403,150 +277,6 @@ Status MysqlResultWriter::append_row_batch(const RowBatch* batch) {
     return status;
 }
 
-Status MysqlResultWriter::append_block(const vectorized::Block& block) {
-    DCHECK(_output_vexpr_ctxs.size() == _result_column_ids.size());
-
-    SCOPED_TIMER(_append_row_batch_timer);
-    if (block.rows() == 0) {
-        return Status::OK();
-    }
-
-    Status status;
-    // convert one batch
-    auto result = std::make_unique<TFetchDataResult>();
-    int num_rows = block.rows();
-    result->result_batch.rows.resize(num_rows);
-
-    for (int i = 0; status.ok() && i < _output_vexpr_ctxs.size(); ++i) {
-        auto column_ptr = block.getByPosition(_result_column_ids[i]).column;
-        auto type_ptr = block.getByPosition(_result_column_ids[i]).type;
-
-        switch (_output_vexpr_ctxs[i]->root()->result_type()) {
-            case TYPE_BOOLEAN:
-            case TYPE_TINYINT:{
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_TINYINT, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_TINYINT, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_SMALLINT: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_SMALLINT, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_SMALLINT, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_INT: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_INT, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_INT, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_BIGINT: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_BIGINT, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_BIGINT, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_LARGEINT: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_LARGEINT, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_LARGEINT, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_FLOAT: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_FLOAT, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_FLOAT, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_DOUBLE: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_DOUBLE, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_DOUBLE, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_CHAR:
-            case TYPE_VARCHAR: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_VARCHAR, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_VARCHAR, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_DECIMALV2: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_DECIMALV2, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_DECIMALV2, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_DATE:
-            case TYPE_DATETIME: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_DATETIME, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_DATETIME, false>(column_ptr);
-                }
-                break;
-            }
-            case TYPE_HLL:
-            case TYPE_OBJECT: {
-                if (type_ptr->isNullable()) {
-                    status = _add_one_column<PrimitiveType::TYPE_OBJECT, true>(column_ptr);
-                } else {
-                    status = _add_one_column<PrimitiveType::TYPE_OBJECT, false>(column_ptr);
-                }
-                break;
-            }
-            default: {
-                LOG(WARNING) << "can't convert this type to mysql type. type = "
-                             << _output_expr_ctxs[i]->root()->type();
-                return Status::InternalError("vec block pack mysql buffer failed.");
-            }
-        }
-
-        if (status.ok()) {
-            for (int j = 0; j < num_rows; ++j) {
-                result->result_batch.rows[j].append(_vec_buffers[j]->buf(), _vec_buffers[j]->length());
-            }
-        } else {
-            LOG(WARNING) << "convert row to mysql result failed.";
-            break;
-        }
-    }
-
-    if (status.ok()) {
-        SCOPED_TIMER(_result_send_timer);
-        // push this batch to back
-        status = _sinker->add_batch(result.get());
-
-        if (status.ok()) {
-            result.release();
-            _written_rows += num_rows;
-        } else {
-            LOG(WARNING) << "append result batch to sink failed.";
-        }
-    }
-
-    return status;
-}
-
 Status MysqlResultWriter::close() {
     COUNTER_SET(_sent_rows_counter, _written_rows);
     return Status::OK();
diff --git a/be/src/runtime/mysql_result_writer.h b/be/src/runtime/mysql_result_writer.h
index 61ba23e6975868..00aea359709fc0 100644
--- a/be/src/runtime/mysql_result_writer.h
+++ b/be/src/runtime/mysql_result_writer.h
@@ -40,7 +40,7 @@ namespace vectorized {
 class MysqlResultWriter final : public ResultWriter {
 public:
     MysqlResultWriter(BufferControlBlock* sinker, const std::vector<ExprContext*>& output_expr_ctxs,
-            const std::vector<vectorized::VExprContext*>& output_vexpr_ctxs, RuntimeProfile* parent_profile);
+            RuntimeProfile* parent_profile);
 
     virtual ~MysqlResultWriter();
 
@@ -49,8 +49,6 @@ class MysqlResultWriter final : public ResultWriter {
     // append this batch to the result sink
     virtual Status append_row_batch(const RowBatch* batch) override;
 
-    virtual Status append_block(const vectorized::Block& block);
-
     virtual Status close() override;
 
 private:
@@ -59,14 +57,10 @@ class MysqlResultWriter final : public ResultWriter {
     Status _add_one_row(TupleRow* row);
     int _add_row_value(int index, const TypeDescriptor& type, void* item);
 
-    template <PrimitiveType type, bool is_nullable>
-    Status _add_one_column(const vectorized::ColumnPtr& column_ptr);
-
 private:
     BufferControlBlock* _sinker;
     const std::vector<ExprContext*>& _output_expr_ctxs;
 
-    const std::vector<vectorized::VExprContext*>& _output_vexpr_ctxs;
     std::vector<int> _result_column_ids;
 
     MysqlRowBuffer* _row_buffer;
diff --git a/be/src/runtime/result_sink.cpp b/be/src/runtime/result_sink.cpp
index 4e4ac5e7f43ff2..563411038277af 100644
--- a/be/src/runtime/result_sink.cpp
+++ b/be/src/runtime/result_sink.cpp
@@ -34,8 +34,8 @@
 namespace doris {
 
 ResultSink::ResultSink(const RowDescriptor& row_desc, const std::vector<TExpr>& t_output_expr,
-                       const TResultSink& sink, int buffer_size, bool is_vec)
-        : _row_desc(row_desc), _t_output_expr(t_output_expr), _buf_size(buffer_size), _is_vec(is_vec) {
+                       const TResultSink& sink, int buffer_size)
+        : _row_desc(row_desc), _t_output_expr(t_output_expr), _buf_size(buffer_size) {
     if (!sink.__isset.type || sink.type == TResultSinkType::MYSQL_PROTOCAL) {
         _sink_type = TResultSinkType::MYSQL_PROTOCAL;
     } else {
@@ -53,17 +53,10 @@ ResultSink::ResultSink(const RowDescriptor& row_desc, const std::vector<TExpr>&
 ResultSink::~ResultSink() {}
 
 Status ResultSink::prepare_exprs(RuntimeState* state) {
-    if (_is_vec) {
-        // From the thrift expressions create the real exprs.
-        RETURN_IF_ERROR(vectorized::VExpr::create_expr_trees(state->obj_pool(), _t_output_expr, &_output_vexpr_ctxs));
-        // Prepare the exprs to run.
-        RETURN_IF_ERROR(vectorized::VExpr::prepare(_output_vexpr_ctxs, state, _row_desc, _expr_mem_tracker));
-    } else {
-        // From the thrift expressions create the real exprs.
-        RETURN_IF_ERROR(Expr::create_expr_trees(state->obj_pool(), _t_output_expr, &_output_expr_ctxs));
-        // Prepare the exprs to run.
-        RETURN_IF_ERROR(Expr::prepare(_output_expr_ctxs, state, _row_desc, _expr_mem_tracker));
-    }
+    // From the thrift expressions create the real exprs.
+    RETURN_IF_ERROR(Expr::create_expr_trees(state->obj_pool(), _t_output_expr, &_output_expr_ctxs));
+    // Prepare the exprs to run.
+    RETURN_IF_ERROR(Expr::prepare(_output_expr_ctxs, state, _row_desc, _expr_mem_tracker));
     return Status::OK();
 }
 
@@ -85,7 +78,7 @@ Status ResultSink::prepare(RuntimeState* state) {
     switch (_sink_type) {
     case TResultSinkType::MYSQL_PROTOCAL:
         _writer.reset(new (std::nothrow)
-                              MysqlResultWriter(_sender.get(), _output_expr_ctxs, _output_vexpr_ctxs, _profile));
+                              MysqlResultWriter(_sender.get(), _output_expr_ctxs, _profile));
         break;
     case TResultSinkType::FILE:
         CHECK(_file_opts.get() != nullptr);
@@ -101,9 +94,6 @@ Status ResultSink::prepare(RuntimeState* state) {
 }
 
 Status ResultSink::open(RuntimeState* state) {
-    if (_is_vec) {
-        return vectorized::VExpr::open(_output_vexpr_ctxs, state);
-    }
     return Expr::open(_output_expr_ctxs, state);
 }
 
@@ -135,11 +125,7 @@ Status ResultSink::close(RuntimeState* state, Status exec_status) {
             time(NULL) + config::result_buffer_cancelled_interval_time,
             state->fragment_instance_id());
 
-    if (_is_vec) {
-        vectorized::VExpr::close(_output_vexpr_ctxs, state);
-    } else {
-        Expr::close(_output_expr_ctxs, state);
-    }
+    Expr::close(_output_expr_ctxs, state);
 
     _closed = true;
     return Status::OK();
diff --git a/be/src/runtime/result_sink.h b/be/src/runtime/result_sink.h
index a96610393b7e84..a7c1da4c51abc1 100644
--- a/be/src/runtime/result_sink.h
+++ b/be/src/runtime/result_sink.h
@@ -46,7 +46,7 @@ class ResultSink : public DataSink {
     // row_desc used for convert RowBatch to TRowBatch
     // buffer_size is the buffer size allocated to each query
     ResultSink(const RowDescriptor& row_desc, const std::vector<TExpr>& select_exprs,
-               const TResultSink& sink, int buffer_size, bool is_vec = false);
+               const TResultSink& sink, int buffer_size);
     virtual ~ResultSink();
     virtual Status prepare(RuntimeState* state);
     virtual Status open(RuntimeState* state);
@@ -73,7 +73,6 @@ class ResultSink : public DataSink {
     // Owned by the RuntimeState.
     const std::vector<TExpr>& _t_output_expr;
     std::vector<ExprContext*> _output_expr_ctxs;
-    std::vector<vectorized::VExprContext*> _output_vexpr_ctxs;
 
     boost::shared_ptr<BufferControlBlock> _sender;
     boost::shared_ptr<ResultWriter> _writer;
diff --git a/be/src/runtime/result_writer.h b/be/src/runtime/result_writer.h
index 0c3b59c2837c86..3c914d06b2c5c0 100644
--- a/be/src/runtime/result_writer.h
+++ b/be/src/runtime/result_writer.h
@@ -34,7 +34,7 @@ namespace vectorized {
 // abstract class of the result writer
 class ResultWriter {
 public:
-    ResultWriter(bool is_vec = false):_is_vec(is_vec) {};
+    ResultWriter() {};
     ~ResultWriter(){};
 
     virtual Status init(RuntimeState* state) = 0;
@@ -53,7 +53,6 @@ class ResultWriter {
 
 protected:
     int64_t _written_rows = 0; // number of rows written
-    bool _is_vec;
 };
 
 } // namespace doris
diff --git a/be/src/vec/exprs/vectorized_agg_fn.cpp b/be/src/vec/exprs/vectorized_agg_fn.cpp
index 7c5fe03516c724..f3c257d99ff83b 100644
--- a/be/src/vec/exprs/vectorized_agg_fn.cpp
+++ b/be/src/vec/exprs/vectorized_agg_fn.cpp
@@ -21,6 +21,8 @@
 #include "fmt/ranges.h"
 #include "runtime/descriptors.h"
 #include "vec/aggregate_functions/aggregate_function_simple_factory.h"
+#include "vec/columns/column_nullable.h"
+#include "vec/data_types/data_type_nullable.h"
 #include "vec/exprs/vexpr.h"
 
 namespace doris::vectorized {
@@ -71,7 +73,10 @@ Status AggFnEvaluator::prepare(RuntimeState* state, const RowDescriptor& desc, M
     doris::vectorized::Array params;
     // prepare for argument
     for (int i = 0; i < _input_exprs_ctxs.size(); ++i) {
-        argument_types.emplace_back(_input_exprs_ctxs[i]->root()->data_type());
+        // Now, For correctness. We have to treat each AggFn argument as nullable. which will cause execute slowly
+        // TODO: RECHCK THE BEHAVIOR
+        auto data_type = _input_exprs_ctxs[i]->root()->data_type();
+        argument_types.emplace_back(data_type->isNullable() ? data_type : std::make_shared<DataTypeNullable>(data_type));
         child_expr_name.emplace_back(_input_exprs_ctxs[i]->root()->expr_name());
     }
     _function = AggregateFunctionSimpleFactory::instance().get(_fn.name.function_name,
@@ -100,12 +105,7 @@ void AggFnEvaluator::destroy(AggregateDataPtr place) {
 }
 
 void AggFnEvaluator::execute_single_add(Block* block, AggregateDataPtr place, Arena* arena) {
-    std::vector<ColumnPtr> columns(_input_exprs_ctxs.size());
-    for (int i = 0; i < _input_exprs_ctxs.size(); ++i) {
-        int column_id = -1;
-        _input_exprs_ctxs[i]->execute(block, &column_id);
-        columns[i] = block->getByPosition(column_id).column->convertToFullColumnIfConst();
-    }
+    auto columns = _get_argment_columns(block);
     // Because the `convertToFullColumnIfConst()` may return a temporary variable, so we need keep the reference of it
     // to make sure program do not destroy it before we call `addBatchSinglePlace`.
     // WARNING:
@@ -119,14 +119,14 @@ void AggFnEvaluator::execute_single_add(Block* block, AggregateDataPtr place, Ar
 
 void AggFnEvaluator::execute_batch_add(Block* block, size_t offset, AggregateDataPtr* places,
                                        Arena* arena) {
-    std::vector<const IColumn*> column_arguments(_input_exprs_ctxs.size());
-    auto columns = block->getColumns();
-    for (int i = 0; i < _input_exprs_ctxs.size(); ++i) {
-        int column_id = -1;
-        _input_exprs_ctxs[i]->execute(block, &column_id);
-        column_arguments[i] =
-                block->getByPosition(column_id).column->convertToFullColumnIfConst().get();
-    }
+    auto columns = _get_argment_columns(block);
+    // Because the `convertToFullColumnIfConst()` may return a temporary variable, so we need keep the reference of it
+    // to make sure program do not destroy it before we call `addBatchSinglePlace`.
+    // WARNING:
+    //      There's danger to call `convertToFullColumnIfConst().get()` to get the `const IColumn*` directly.
+    std::vector<const IColumn*> column_arguments(columns.size());
+    std::transform(columns.cbegin(), columns.cend(), column_arguments.begin(),
+                   [](const auto& ptr) { return ptr.get(); });
     SCOPED_TIMER(_exec_timer);
     _function->addBatch(block->rows(), places, offset, column_arguments.data(), arena);
 }
@@ -158,4 +158,16 @@ std::string AggFnEvaluator::debug_string() const {
     out << ")";
     return out.str();
 }
+
+std::vector<ColumnPtr> AggFnEvaluator::_get_argment_columns(Block* block) const {
+    std::vector<ColumnPtr> columns(_input_exprs_ctxs.size());
+    for (int i = 0; i < _input_exprs_ctxs.size(); ++i) {
+        int column_id = -1;
+        _input_exprs_ctxs[i]->execute(block, &column_id);
+        auto ptr = block->getByPosition(column_id).column->convertToFullColumnIfConst();
+        columns[i] = makeNullable(ptr);
+    }
+    return columns;
+}
+
 } // namespace doris::vectorized
diff --git a/be/src/vec/exprs/vectorized_agg_fn.h b/be/src/vec/exprs/vectorized_agg_fn.h
index 85e29f6d271fdd..f8df3bdfcd73e0 100644
--- a/be/src/vec/exprs/vectorized_agg_fn.h
+++ b/be/src/vec/exprs/vectorized_agg_fn.h
@@ -71,6 +71,8 @@ class AggFnEvaluator {
 
     AggFnEvaluator(const TExprNode& desc);
 
+    std::vector<ColumnPtr> _get_argment_columns(Block* block) const;
+
     const TypeDescriptor _return_type;
     const TypeDescriptor _intermediate_type;
 
diff --git a/be/src/vec/sink/mysql_result_writer.cpp b/be/src/vec/sink/mysql_result_writer.cpp
index 7e5942116c5899..dcbc8aea3f3310 100644
--- a/be/src/vec/sink/mysql_result_writer.cpp
+++ b/be/src/vec/sink/mysql_result_writer.cpp
@@ -31,7 +31,7 @@ namespace vectorized {
 MysqlResultWriter::MysqlResultWriter(BufferControlBlock* sinker,
                                      const std::vector<VExprContext*>& output_vexpr_ctxs,
                                      RuntimeProfile* parent_profile)
-        : VResultWriter(!output_vexpr_ctxs.empty()),
+        : VResultWriter(),
           _sinker(sinker),
           _output_vexpr_ctxs(output_vexpr_ctxs),
           _parent_profile(parent_profile) {}
@@ -48,12 +48,11 @@ Status MysqlResultWriter::init(RuntimeState* state) {
         return Status::InternalError("sinker is NULL pointer.");
     }
 
-    if (_is_vec) {
-        _vec_buffers.resize(state->batch_size());
-        for (int i = 0; i < state->batch_size(); ++i) {
-            _vec_buffers[i] = new MysqlRowBuffer();
-        }
+    _vec_buffers.resize(state->batch_size());
+    for (int i = 0; i < state->batch_size(); ++i) {
+        _vec_buffers[i] = new MysqlRowBuffer();
     }
+
     return Status::OK();
 }
 
diff --git a/be/src/vec/sink/result_writer.h b/be/src/vec/sink/result_writer.h
index bf9bebbc062bb8..ef03980007b964 100644
--- a/be/src/vec/sink/result_writer.h
+++ b/be/src/vec/sink/result_writer.h
@@ -21,7 +21,7 @@ namespace doris {
 namespace vectorized {
 class VResultWriter : public ResultWriter {
 public:
-    VResultWriter(bool is_vec) : ResultWriter(is_vec) {}
+    VResultWriter() : ResultWriter() {}
 
     virtual Status append_block(Block& block) = 0;
 };
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Function.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Function.java
index 6c6b98eecbd177..c85226370a25de 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Function.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Function.java
@@ -146,6 +146,22 @@ public Function(long id, FunctionName name, Type[] argTypes, Type retType, boole
         this.vectorized = vectorized;
     }
 
+    public Function(long id, FunctionName name, Type[] argTypes, Type retType, boolean hasVarArgs,
+                    TFunctionBinaryType binaryType, boolean userVisible, boolean vectorized) {
+        this.id = id;
+        this.name = name;
+        this.hasVarArgs = hasVarArgs;
+        if (argTypes == null) {
+            this.argTypes = new Type[0];
+        } else {
+            this.argTypes = argTypes;
+        }
+        this.retType = retType;
+        this.binaryType = binaryType;
+        this.userVisible = userVisible;
+        this.vectorized = vectorized;
+    }
+
     public Function(long id, FunctionName name, List<Type> argTypes, Type retType, boolean hasVarArgs, boolean vectorized) {
         this(id, name, (Type[]) null, retType, hasVarArgs, vectorized);
         if (argTypes.size() > 0) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java
index 2e3c0203a7ffe2..4c43a1d5a9e32d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java
@@ -1036,7 +1036,8 @@ public void addBuiltinBothScalaAndVectorized(Function fn) {
             vecFns = Lists.newArrayList();
             vectorizedFunctions.put(fn.functionName(), vecFns);
         }
-        vecFns.add(new Function(fn.getFunctionName(), fn.getArgs(), fn.getReturnType(), fn.hasVarArgs(), true));
+        vecFns.add(new Function(fn.getId(), fn.getFunctionName(), fn.getArgs(), fn.getReturnType(), fn.hasVarArgs(),
+                fn.getBinaryType(), fn.isUserVisible(), true));
     }