diff --git a/be/src/vec/columns/column_nullable.h b/be/src/vec/columns/column_nullable.h
index 4350bead62d131..d183ce03853026 100644
--- a/be/src/vec/columns/column_nullable.h
+++ b/be/src/vec/columns/column_nullable.h
@@ -134,6 +134,16 @@ class ColumnNullable final : public COWHelper<IColumn, ColumnNullable> {
     void insert(const Field& x) override;
     void insert_from(const IColumn& src, size_t n) override;
 
+    template <typename ColumnType>
+    void insert_from_with_type(const IColumn& src, size_t n) {
+        const ColumnNullable& src_concrete = assert_cast<const ColumnNullable&>(src);
+        assert_cast<ColumnType*>(nested_column.get())
+                ->insert_from(src_concrete.get_nested_column(), n);
+        auto is_null = src_concrete.get_null_map_data()[n];
+        _has_null |= is_null;
+        _get_null_map_data().push_back(is_null);
+    }
+
     void insert_from_not_nullable(const IColumn& src, size_t n);
     void insert_range_from_not_nullable(const IColumn& src, size_t start, size_t length);
     void insert_many_from_not_nullable(const IColumn& src, size_t position, size_t length);
diff --git a/be/src/vec/functions/function_case.h b/be/src/vec/functions/function_case.h
index 944d61ac43af9c..ca59568aecdd20 100644
--- a/be/src/vec/functions/function_case.h
+++ b/be/src/vec/functions/function_case.h
@@ -31,6 +31,7 @@
 #include "vec/aggregate_functions/aggregate_function.h"
 #include "vec/columns/column.h"
 #include "vec/columns/column_complex.h"
+#include "vec/columns/column_const.h"
 #include "vec/columns/column_nullable.h"
 #include "vec/columns/columns_number.h"
 #include "vec/core/block.h"
@@ -197,7 +198,8 @@ class FunctionCase : public IFunction {
         }
 
         auto result_column_ptr = data_type->create_column();
-        update_result_normal(result_column_ptr, then_idx, column_holder);
+        update_result_normal<int, ColumnType, then_null>(result_column_ptr, then_idx,
+                                                         column_holder);
         block.replace_by_position(result, std::move(result_column_ptr));
         return Status::OK();
     }
@@ -269,11 +271,13 @@ class FunctionCase : public IFunction {
                       std::is_same_v<ColumnType, ColumnHLL>) {
             // result_column and all then_column is not nullable.
             // can't simd when type is string.
-            update_result_normal(result_column_ptr, then_idx, column_holder);
+            update_result_normal<uint8_t, ColumnType, then_null>(result_column_ptr, then_idx,
+                                                                 column_holder);
         } else if constexpr (then_null) {
             // result_column and all then_column is nullable.
             // TODO: make here simd automatically.
-            update_result_normal(result_column_ptr, then_idx, column_holder);
+            update_result_normal<uint8_t, ColumnType, then_null>(result_column_ptr, then_idx,
+                                                                 column_holder);
         } else {
             update_result_auto_simd<ColumnType>(result_column_ptr, then_idx, column_holder);
         }
@@ -282,9 +286,17 @@ class FunctionCase : public IFunction {
         return Status::OK();
     }
 
-    template <typename IndexType>
+    template <typename IndexType, typename ColumnType, bool then_null>
     void update_result_normal(MutableColumnPtr& result_column_ptr, IndexType* then_idx,
                               CaseWhenColumnHolder& column_holder) {
+        std::vector<uint8_t> is_consts(column_holder.then_ptrs.size());
+        std::vector<ColumnPtr> raw_columns(column_holder.then_ptrs.size());
+        for (size_t i = 0; i < column_holder.then_ptrs.size(); i++) {
+            if (column_holder.then_ptrs[i].has_value()) {
+                std::tie(raw_columns[i], is_consts[i]) =
+                        unpack_if_const(column_holder.then_ptrs[i].value());
+            }
+        }
         for (int row_idx = 0; row_idx < column_holder.rows_count; row_idx++) {
             if constexpr (!has_else) {
                 if (!then_idx[row_idx]) {
@@ -292,8 +304,15 @@ class FunctionCase : public IFunction {
                     continue;
                 }
             }
-            result_column_ptr->insert_from(*column_holder.then_ptrs[then_idx[row_idx]].value(),
-                                           row_idx);
+            size_t target = is_consts[then_idx[row_idx]] ? 0 : row_idx;
+            if constexpr (then_null) {
+                assert_cast<ColumnNullable*>(result_column_ptr.get())
+                        ->insert_from_with_type<ColumnType>(*raw_columns[then_idx[row_idx]],
+                                                            target);
+            } else {
+                assert_cast<ColumnType*>(result_column_ptr.get())
+                        ->insert_from(*raw_columns[then_idx[row_idx]], target);
+            }
         }
     }
 
@@ -301,6 +320,11 @@ class FunctionCase : public IFunction {
     void update_result_auto_simd(MutableColumnPtr& result_column_ptr,
                                  const uint8* __restrict then_idx,
                                  CaseWhenColumnHolder& column_holder) {
+        for (size_t i = 0; i < column_holder.then_ptrs.size(); i++) {
+            column_holder.then_ptrs[i]->reset(
+                    column_holder.then_ptrs[i].value()->convert_to_full_column_if_const());
+        }
+
         size_t rows_count = column_holder.rows_count;
         result_column_ptr->resize(rows_count);
         auto* __restrict result_raw_data =
@@ -361,20 +385,18 @@ class FunctionCase : public IFunction {
                                  size_t input_rows_count) {
         bool when_null = false;
         if constexpr (has_case) {
+            block.replace_by_position_if_const(arguments[0]);
             if (block.get_by_position(arguments[0]).type->is_nullable()) {
                 when_null = true;
             }
         }
         for (int i = has_case; i < arguments.size() - has_else; i += 2) {
+            block.replace_by_position_if_const(arguments[i]);
             if (block.get_by_position(arguments[i]).type->is_nullable()) {
                 when_null = true;
             }
         }
 
-        for (int i = 0; i < arguments.size(); i++) {
-            block.replace_by_position_if_const(arguments[i]);
-        }
-
         if (when_null) {
             return execute_get_then_null<ColumnType, true>(data_type, block, arguments, result,
                                                            input_rows_count);