diff --git a/be/src/exprs/function/array/function_array_apply.cpp b/be/src/exprs/function/array/function_array_apply.cpp
index ddbaa3b1c6495e..57a1a8ad1ee9b1 100644
--- a/be/src/exprs/function/array/function_array_apply.cpp
+++ b/be/src/exprs/function/array/function_array_apply.cpp
@@ -33,8 +33,9 @@
 #include "core/call_on_type_index.h"
 #include "core/column/column.h"
 #include "core/column/column_array.h"
+#include "core/column/column_array_view.h"
 #include "core/column/column_const.h"
-#include "core/column/column_nullable.h"
+#include "core/column/column_decimal.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/data_type.h"
 #include "core/data_type/data_type_array.h"
@@ -82,7 +83,6 @@ class FunctionArrayApply : public IFunction {
                     fmt::format("unsupported types for function {}({})", get_name(),
                                 block.get_by_position(arguments[0]).type->get_name()));
         }
-        const auto& src_offsets = src_column_array->get_offsets();
         const auto* src_nested_column = &src_column_array->get_data();
         DCHECK(src_nested_column != nullptr);
 
@@ -95,7 +95,7 @@ class FunctionArrayApply : public IFunction {
                 static_cast<const ColumnConst&>(*block.get_by_position(arguments[2]).column.get());
         ColumnPtr result_ptr;
         RETURN_IF_CATCH_EXCEPTION(
-                RETURN_IF_ERROR(_execute(*src_nested_column, nested_type, src_offsets, condition,
+                RETURN_IF_ERROR(_execute(src_column, *src_nested_column, nested_type, condition,
                                          rhs_value_column, &result_ptr)));
         block.replace_by_position(result, std::move(result_ptr));
         return Status::OK();
@@ -135,37 +135,30 @@ class FunctionArrayApply : public IFunction {
     }
 
     // need exception safety
-    template <typename T, ApplyOp op>
-    ColumnPtr _apply_internal(const IColumn& src_column, const ColumnArray::Offsets64& src_offsets,
+    template <PrimitiveType PType, ApplyOp op>
+    ColumnPtr _apply_internal(const ColumnArrayView<PType>& array_view, const IColumn& src_column,
                               const ColumnConst& cmp) const {
+        using T = typename PrimitiveTypeTraits<PType>::CppType;
         T rhs_val = *reinterpret_cast<const T*>(cmp.get_data_at(0).data);
         auto column_filter = ColumnUInt8::create(src_column.size(), 0);
         auto& column_filter_data = column_filter->get_data();
-        const char* src_column_data_ptr = nullptr;
-        const uint8_t* null_map_data = nullptr;
-        if (!is_column_nullable(src_column)) {
-            src_column_data_ptr = src_column.get_raw_data().data;
-        } else {
-            const auto* nullable_col = assert_cast<const ColumnNullable*>(&src_column);
-            src_column_data_ptr = nullable_col->get_nested_column().get_raw_data().data;
-            null_map_data = nullable_col->get_null_map_data().data();
-        }
-        const T* src_column_data_t_ptr = reinterpret_cast<const T*>(src_column_data_ptr);
+        const T* src_column_data_t_ptr = reinterpret_cast<const T*>(array_view.get_data());
+        const UInt8* null_map_data = array_view.get_null_map_data();
         const size_t src_column_size = src_column.size();
         for (size_t i = 0; i < src_column_size; ++i) {
-            if (null_map_data && null_map_data[i]) {
+            if (null_map_data[i]) {
                 continue; // null elements should not pass the filter
             }
             column_filter_data[i] = apply<T, op>(src_column_data_t_ptr[i], rhs_val);
         }
         const IColumn::Filter& filter = column_filter_data;
         ColumnPtr filtered = src_column.filter(filter, src_column.size());
-        auto column_offsets = ColumnArray::ColumnOffsets::create(src_offsets.size());
+        auto column_offsets = ColumnArray::ColumnOffsets::create(array_view.offsets.size());
         ColumnArray::Offsets64& dst_offsets = column_offsets->get_data();
         size_t in_pos = 0;
         size_t out_pos = 0;
-        for (size_t i = 0; i < src_offsets.size(); ++i) {
-            for (; in_pos < src_offsets[i]; ++in_pos) {
+        for (size_t i = 0; i < array_view.offsets.size(); ++i) {
+            for (; in_pos < array_view.offsets[i]; ++in_pos) {
                 if (filter[in_pos]) {
                     ++out_pos;
                 }
@@ -176,14 +169,14 @@ class FunctionArrayApply : public IFunction {
     }
 
     template <ApplyOp OP>
-    void dispatch_array_scalar(DataTypePtr nested_type, const IColumn& src_column,
-                               const ColumnArray::Offsets64& src_offsets, const ColumnConst& cmp,
+    void dispatch_array_scalar(const ColumnPtr& array_column, DataTypePtr nested_type,
+                               const IColumn& src_column, const ColumnConst& cmp,
                                ColumnPtr* dst) const {
         auto call = [&](const auto& type) -> bool {
             using DispatchType = std::decay_t<decltype(type)>;
             constexpr PrimitiveType PType = DispatchType::PType;
-            *dst = _apply_internal<typename PrimitiveTypeTraits<PType>::CppType, OP>(
-                    src_column, src_offsets, cmp);
+            auto array_view = ColumnArrayView<PType>::create(array_column);
+            *dst = _apply_internal<PType, OP>(array_view, src_column, cmp);
             return true;
         };
 
@@ -195,27 +188,27 @@ class FunctionArrayApply : public IFunction {
         }
     }
     // need exception safety
-    Status _execute(const IColumn& nested_src, DataTypePtr nested_type,
-                    const ColumnArray::Offsets64& offsets, const std::string& condition,
+    Status _execute(const ColumnPtr& array_column, const IColumn& nested_src,
+                    DataTypePtr nested_type, const std::string& condition,
                     const ColumnConst& rhs_value_column, ColumnPtr* dst) const {
         if (condition == "=") {
-            dispatch_array_scalar<ApplyOp::EQ>(nested_type, nested_src, offsets, rhs_value_column,
-                                               dst);
+            dispatch_array_scalar<ApplyOp::EQ>(array_column, nested_type, nested_src,
+                                               rhs_value_column, dst);
         } else if (condition == "!=") {
-            dispatch_array_scalar<ApplyOp::NE>(nested_type, nested_src, offsets, rhs_value_column,
-                                               dst);
+            dispatch_array_scalar<ApplyOp::NE>(array_column, nested_type, nested_src,
+                                               rhs_value_column, dst);
         } else if (condition == "<") {
-            dispatch_array_scalar<ApplyOp::LT>(nested_type, nested_src, offsets, rhs_value_column,
-                                               dst);
+            dispatch_array_scalar<ApplyOp::LT>(array_column, nested_type, nested_src,
+                                               rhs_value_column, dst);
         } else if (condition == "<=") {
-            dispatch_array_scalar<ApplyOp::LE>(nested_type, nested_src, offsets, rhs_value_column,
-                                               dst);
+            dispatch_array_scalar<ApplyOp::LE>(array_column, nested_type, nested_src,
+                                               rhs_value_column, dst);
         } else if (condition == ">") {
-            dispatch_array_scalar<ApplyOp::GT>(nested_type, nested_src, offsets, rhs_value_column,
-                                               dst);
+            dispatch_array_scalar<ApplyOp::GT>(array_column, nested_type, nested_src,
+                                               rhs_value_column, dst);
         } else if (condition == ">=") {
-            dispatch_array_scalar<ApplyOp::GE>(nested_type, nested_src, offsets, rhs_value_column,
-                                               dst);
+            dispatch_array_scalar<ApplyOp::GE>(array_column, nested_type, nested_src,
+                                               rhs_value_column, dst);
         } else {
             return Status::RuntimeError(
                     fmt::format("execute failed, unsupported op {} for function {})", condition,
diff --git a/be/src/exprs/function/array/function_array_count.cpp b/be/src/exprs/function/array/function_array_count.cpp
index 98d65d000c05e4..cd7b9440c2c414 100644
--- a/be/src/exprs/function/array/function_array_count.cpp
+++ b/be/src/exprs/function/array/function_array_count.cpp
@@ -15,9 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include "core/column/column_array.h"
-#include "core/column/column_nullable.h"
-#include "core/data_type/data_type_array.h"
+#include "core/column/column_array_view.h"
 #include "core/data_type/data_type_number.h"
 #include "exprs/function/function.h"
 #include "exprs/function/function_helpers.h"
@@ -48,52 +46,25 @@ class FunctionArrayCount : public IFunction {
 
     Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments,
                         uint32_t result, size_t input_rows_count) const override {
-        const auto& [src_column, src_const] =
-                unpack_if_const(block.get_by_position(arguments[0]).column);
-        const ColumnArray* array_column = nullptr;
-        const UInt8* array_null_map = nullptr;
-        if (const auto* nullable_array = check_and_get_column<ColumnNullable>(src_column.get())) {
-            array_column = assert_cast<const ColumnArray*>(&nullable_array->get_nested_column());
-            array_null_map = nullable_array->get_null_map_column().get_data().data();
-        } else {
-            array_column = assert_cast<const ColumnArray*>(src_column.get());
-        }
-
-        if (!array_column) {
-            return Status::RuntimeError("unsupported types for function {}({})", get_name(),
-                                        block.get_by_position(arguments[0]).type->get_name());
-        }
-
-        const auto& offsets = array_column->get_offsets();
-        ColumnPtr nested_column = nullptr;
-        const UInt8* nested_null_map = nullptr;
-        if (is_column_nullable(array_column->get_data())) {
-            const auto& nested_null_column =
-                    assert_cast<const ColumnNullable&>(array_column->get_data());
-            nested_null_map = nested_null_column.get_null_map_column().get_data().data();
-            nested_column = nested_null_column.get_nested_column_ptr();
-        } else {
-            nested_column = array_column->get_data_ptr();
-        }
-
-        const auto& nested_data = assert_cast<const ColumnUInt8&>(*nested_column).get_data();
-
-        auto dst_column = ColumnInt64::create(offsets.size());
+        auto array_view =
+                ColumnArrayView<TYPE_BOOLEAN>::create(block.get_by_position(arguments[0]).column);
+        auto dst_column = ColumnInt64::create(array_view.size());
         auto& dst_data = dst_column->get_data();
 
-        for (size_t row = 0; row < offsets.size(); ++row) {
+        for (size_t row = 0; row < array_view.size(); ++row) {
             Int64 res = 0;
-            if (array_null_map && array_null_map[row]) {
+            if (array_view.is_null_at(row)) {
                 dst_data[row] = res;
                 continue;
             }
-            size_t off = offsets[row - 1];
-            size_t len = offsets[row] - off;
-            for (size_t pos = 0; pos < len; ++pos) {
-                if (nested_null_map && nested_null_map[pos + off]) {
+            auto array_data = array_view[row];
+            const auto* data = array_data.get_data();
+            const auto* null_map = array_data.get_null_map_data();
+            for (size_t pos = 0; pos < array_data.size(); ++pos) {
+                if (null_map[pos]) {
                     continue;
                 }
-                if (nested_data[pos + off] != 0) {
+                if (data[pos] != 0) {
                     ++res;
                 }
             }
diff --git a/be/src/exprs/function/array/function_array_cum_sum.cpp b/be/src/exprs/function/array/function_array_cum_sum.cpp
index d5dc5436ad851c..2789071ea7cc0c 100644
--- a/be/src/exprs/function/array/function_array_cum_sum.cpp
+++ b/be/src/exprs/function/array/function_array_cum_sum.cpp
@@ -23,6 +23,7 @@
 #include "core/call_on_type_index.h"
 #include "core/column/column.h"
 #include "core/column/column_array.h"
+#include "core/column/column_array_view.h"
 #include "core/data_type/data_type.h"
 #include "core/data_type/data_type_array.h"
 #include "core/data_type/data_type_decimal.h"
@@ -125,21 +126,10 @@ class FunctionArrayCumSum : public IFunction {
                                 block.get_by_position(arguments[0]).type->get_name()));
         }
 
-        const auto& src_offsets = src_column_array->get_offsets();
-        const auto* src_nested_column = &src_column_array->get_data();
-        DCHECK(src_nested_column != nullptr);
-
-        // get src nested column
         auto src_nested_type = assert_cast<const DataTypeArray&>(*src_arg.type).get_nested_type();
 
-        // get null map
-        const auto* src_nested_nullable_col = assert_cast<const ColumnNullable*>(src_nested_column);
-        src_nested_column = src_nested_nullable_col->get_nested_column_ptr().get();
-        const NullMapType& src_null_map = src_nested_nullable_col->get_null_map_column().get_data();
-
         ColumnPtr res_nested_ptr;
-        auto res_val = _execute_by_type(src_nested_type, *src_nested_column, src_offsets,
-                                        src_null_map, res_nested_ptr);
+        auto res_val = _execute_by_type(src_nested_type, src_column, res_nested_ptr);
         if (!res_val) {
             return Status::InvalidArgument(
                     "execute failed or unsupported types for function {}({})", get_name(),
@@ -154,62 +144,61 @@ class FunctionArrayCumSum : public IFunction {
     }
 
 private:
-    bool _execute_by_type(DataTypePtr src_nested_type, const IColumn& src_column,
-                          const ColumnArray::Offsets64& src_offsets,
-                          const NullMapType& src_null_map, ColumnPtr& res_nested_ptr) const {
+    bool _execute_by_type(DataTypePtr src_nested_type, const ColumnPtr& src_column,
+                          ColumnPtr& res_nested_ptr) const {
         bool res = false;
         switch (src_nested_type->get_primitive_type()) {
         case TYPE_BOOLEAN:
-            res = _execute_number<TYPE_BOOLEAN, TYPE_BIGINT>(src_column, src_offsets, src_null_map,
-                                                             res_nested_ptr);
+            res = _execute_number<TYPE_BOOLEAN, TYPE_BIGINT>(
+                    ColumnArrayView<TYPE_BOOLEAN>::create(src_column), res_nested_ptr);
             break;
         case TYPE_TINYINT:
-            res = _execute_number<TYPE_TINYINT, TYPE_BIGINT>(src_column, src_offsets, src_null_map,
-                                                             res_nested_ptr);
+            res = _execute_number<TYPE_TINYINT, TYPE_BIGINT>(
+                    ColumnArrayView<TYPE_TINYINT>::create(src_column), res_nested_ptr);
             break;
         case TYPE_SMALLINT:
-            res = _execute_number<TYPE_SMALLINT, TYPE_BIGINT>(src_column, src_offsets, src_null_map,
-                                                              res_nested_ptr);
+            res = _execute_number<TYPE_SMALLINT, TYPE_BIGINT>(
+                    ColumnArrayView<TYPE_SMALLINT>::create(src_column), res_nested_ptr);
             break;
         case TYPE_INT:
-            res = _execute_number<TYPE_INT, TYPE_BIGINT>(src_column, src_offsets, src_null_map,
-                                                         res_nested_ptr);
+            res = _execute_number<TYPE_INT, TYPE_BIGINT>(
+                    ColumnArrayView<TYPE_INT>::create(src_column), res_nested_ptr);
             break;
         case TYPE_BIGINT:
-            res = _execute_number<TYPE_BIGINT, TYPE_BIGINT>(src_column, src_offsets, src_null_map,
-                                                            res_nested_ptr);
+            res = _execute_number<TYPE_BIGINT, TYPE_BIGINT>(
+                    ColumnArrayView<TYPE_BIGINT>::create(src_column), res_nested_ptr);
             break;
         case TYPE_LARGEINT:
-            res = _execute_number<TYPE_LARGEINT, TYPE_LARGEINT>(src_column, src_offsets,
-                                                                src_null_map, res_nested_ptr);
+            res = _execute_number<TYPE_LARGEINT, TYPE_LARGEINT>(
+                    ColumnArrayView<TYPE_LARGEINT>::create(src_column), res_nested_ptr);
             break;
         case TYPE_FLOAT:
-            res = _execute_number<TYPE_FLOAT, TYPE_DOUBLE>(src_column, src_offsets, src_null_map,
-                                                           res_nested_ptr);
+            res = _execute_number<TYPE_FLOAT, TYPE_DOUBLE>(
+                    ColumnArrayView<TYPE_FLOAT>::create(src_column), res_nested_ptr);
             break;
         case TYPE_DOUBLE:
-            res = _execute_number<TYPE_DOUBLE, TYPE_DOUBLE>(src_column, src_offsets, src_null_map,
-                                                            res_nested_ptr);
+            res = _execute_number<TYPE_DOUBLE, TYPE_DOUBLE>(
+                    ColumnArrayView<TYPE_DOUBLE>::create(src_column), res_nested_ptr);
             break;
         case TYPE_DECIMAL32:
-            res = _execute_number<TYPE_DECIMAL32, PType>(src_column, src_offsets, src_null_map,
-                                                         res_nested_ptr);
+            res = _execute_number<TYPE_DECIMAL32, PType>(
+                    ColumnArrayView<TYPE_DECIMAL32>::create(src_column), res_nested_ptr);
             break;
         case TYPE_DECIMAL64:
-            res = _execute_number<TYPE_DECIMAL64, PType>(src_column, src_offsets, src_null_map,
-                                                         res_nested_ptr);
+            res = _execute_number<TYPE_DECIMAL64, PType>(
+                    ColumnArrayView<TYPE_DECIMAL64>::create(src_column), res_nested_ptr);
             break;
         case TYPE_DECIMAL128I:
-            res = _execute_number<TYPE_DECIMAL128I, PType>(src_column, src_offsets, src_null_map,
-                                                           res_nested_ptr);
+            res = _execute_number<TYPE_DECIMAL128I, PType>(
+                    ColumnArrayView<TYPE_DECIMAL128I>::create(src_column), res_nested_ptr);
             break;
         case TYPE_DECIMAL256:
-            res = _execute_number<TYPE_DECIMAL256, PType>(src_column, src_offsets, src_null_map,
-                                                          res_nested_ptr);
+            res = _execute_number<TYPE_DECIMAL256, PType>(
+                    ColumnArrayView<TYPE_DECIMAL256>::create(src_column), res_nested_ptr);
             break;
         case TYPE_DECIMALV2:
-            res = _execute_number<TYPE_DECIMALV2, TYPE_DECIMALV2>(src_column, src_offsets,
-                                                                  src_null_map, res_nested_ptr);
+            res = _execute_number<TYPE_DECIMALV2, TYPE_DECIMALV2>(
+                    ColumnArrayView<TYPE_DECIMALV2>::create(src_column), res_nested_ptr);
             break;
         default:
             break;
@@ -218,42 +207,36 @@ class FunctionArrayCumSum : public IFunction {
     }
 
     template <PrimitiveType Element, PrimitiveType Result>
-    bool _execute_number(const IColumn& src_column, const ColumnArray::Offsets64& src_offsets,
-                         const NullMapType& src_null_map, ColumnPtr& res_nested_ptr) const {
+    bool _execute_number(const ColumnArrayView<Element>& array_view,
+                         ColumnPtr& res_nested_ptr) const {
         if constexpr (is_decimalv3(Element) &&
                       (TYPE_DECIMAL128I != Result && TYPE_DECIMAL256 != Result)) {
             return false;
         } else {
-            using ColVecType = typename PrimitiveTypeTraits<Element>::ColumnType;
             using ColVecResult = typename PrimitiveTypeTraits<Result>::ColumnType;
 
-            // 1. get pod array from src
-            auto src_column_concrete = assert_cast<const ColVecType*>(&src_column);
-            if (!src_column_concrete) {
-                return false;
-            }
-
             // 2. construct result data
             typename ColVecResult::MutablePtr res_nested_mut_ptr = nullptr;
             if constexpr (is_decimal(Result)) {
-                res_nested_mut_ptr = ColVecResult::create(0, src_column_concrete->get_scale());
+                res_nested_mut_ptr =
+                        ColVecResult::create(0, array_view.element_data.data.get_scale());
             } else {
                 res_nested_mut_ptr = ColVecResult::create();
             }
 
             // get result data pod array
-            auto size = src_column.size();
+            auto size = array_view.element_data.size();
             auto& res_datas = res_nested_mut_ptr->get_data();
             res_datas.resize(size);
 
             // 3. compute cum sum and null map
-            _compute_cum_sum<Result>(src_column_concrete->get_data(), src_offsets, src_null_map,
-                                     res_datas);
+            _compute_cum_sum<Result>(array_view.element_data.data, array_view.offsets,
+                                     array_view.nested_null_map, res_datas);
 
             // handle null value in res_datas for first null value
             auto res_null_map_col = ColumnUInt8::create(size, 0);
             size_t first_not_null_pos =
-                    VectorizedUtils::find_first_valid_simd(src_null_map, 0, size);
+                    VectorizedUtils::find_first_valid_simd(array_view.nested_null_map, 0, size);
             VLOG_DEBUG << "first_not_null_pos: " << std::to_string(first_not_null_pos);
             VectorizedUtils::range_set_nullmap_to_true_simd(res_null_map_col->get_data(), 0,
                                                             first_not_null_pos);
@@ -330,4 +313,4 @@ void register_function_array_cum_sum(SimpleFunctionFactory& factory) {
     factory.register_array_agg_function("array_cum_sum", creator);
 }
 
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/src/exprs/function/array/function_array_difference.h b/be/src/exprs/function/array/function_array_difference.h
index cb29ef42737cfb..602d442f57b40b 100644
--- a/be/src/exprs/function/array/function_array_difference.h
+++ b/be/src/exprs/function/array/function_array_difference.h
@@ -35,6 +35,7 @@
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column.h"
 #include "core/column/column_array.h"
+#include "core/column/column_array_view.h"
 #include "core/column/column_decimal.h"
 #include "core/column/column_nullable.h"
 #include "core/column/column_vector.h"
@@ -143,47 +144,39 @@ class FunctionArrayDifference : public IFunction {
     }
 
     template <PrimitiveType Element, PrimitiveType Result>
-    ColumnPtr _execute_number_expanded(const ColumnArray::Offsets64& offsets,
-                                       const IColumn& nested_column,
-                                       ColumnPtr nested_null_map) const {
-        using ColVecType = typename PrimitiveTypeTraits<Element>::ColumnType;
+    ColumnPtr _execute_number_expanded(const ColumnArrayView<Element>& array_view) const {
         using ColVecResult = typename PrimitiveTypeTraits<Result>::ColumnType;
         typename ColVecResult::MutablePtr res_nested = nullptr;
 
-        const auto& src_data = reinterpret_cast<const ColVecType&>(nested_column).get_data();
+        const auto& src_data = array_view.element_data.data;
         if constexpr (is_decimal(Result)) {
             res_nested = ColVecResult::create(0, src_data.get_scale());
         } else {
             res_nested = ColVecResult::create();
         }
-        auto size = nested_column.size();
+        auto size = array_view.element_data.size();
         typename ColVecResult::Container& res_values = res_nested->get_data();
         res_values.resize(size);
 
         size_t pos = 0;
-        for (auto offset : offsets) {
+        for (auto offset : array_view.offsets) {
             impl(src_data.data(), res_values.data(), pos, offset);
             pos = offset;
         }
-        if (nested_null_map) {
-            auto null_map_col = ColumnUInt8::create(size, 0);
-            auto& null_map_col_data = null_map_col->get_data();
-            auto nested_colum_data = static_cast<const ColumnUInt8*>(nested_null_map.get());
-            VectorizedUtils::update_null_map(null_map_col_data, nested_colum_data->get_data());
-            for (size_t row = 0; row < offsets.size(); ++row) {
-                auto off = offsets[row - 1];
-                auto len = offsets[row] - off;
-                auto nested_pos = len ? len - 1 : 0;
-                for (; nested_pos > 0; --nested_pos) {
-                    if (null_map_col_data[nested_pos + off - 1]) {
-                        null_map_col_data[nested_pos + off] = 1;
-                    }
+        auto null_map_col = ColumnUInt8::create(size, 0);
+        auto& null_map_col_data = null_map_col->get_data();
+        VectorizedUtils::update_null_map(null_map_col_data, array_view.nested_null_map);
+        for (size_t row = 0; row < array_view.offsets.size(); ++row) {
+            auto off = array_view.offsets[row - 1];
+            auto len = array_view.offsets[row] - off;
+            auto nested_pos = len ? len - 1 : 0;
+            for (; nested_pos > 0; --nested_pos) {
+                if (null_map_col_data[nested_pos + off - 1]) {
+                    null_map_col_data[nested_pos + off] = 1;
                 }
             }
-            return ColumnNullable::create(std::move(res_nested), std::move(null_map_col));
-        } else {
-            return res_nested;
         }
+        return ColumnNullable::create(std::move(res_nested), std::move(null_map_col));
     }
 
     ColumnPtr _execute_non_nullable(const ColumnWithTypeAndName& arg,
@@ -194,72 +187,61 @@ class FunctionArrayDifference : public IFunction {
         const auto& offsets = array_column.get_offsets();
         DCHECK(offsets.size() == input_rows_count);
 
-        ColumnPtr nested_column = nullptr;
-        ColumnPtr nested_null_map = nullptr;
-        if (is_column_nullable(array_column.get_data())) {
-            const auto& nested_null_column =
-                    reinterpret_cast<const ColumnNullable&>(array_column.get_data());
-            nested_column = nested_null_column.get_nested_column_ptr();
-            nested_null_map = nested_null_column.get_null_map_column_ptr();
-        } else {
-            nested_column = array_column.get_data_ptr();
-        }
-
         ColumnPtr res = nullptr;
         auto left_element_type =
                 remove_nullable(assert_cast<const DataTypeArray&>(*arg.type).get_nested_type());
         switch (left_element_type->get_primitive_type()) {
         case TYPE_BOOLEAN:
-            res = _execute_number_expanded<TYPE_BOOLEAN, TYPE_SMALLINT>(offsets, *nested_column,
-                                                                        nested_null_map);
+            res = _execute_number_expanded<TYPE_BOOLEAN, TYPE_SMALLINT>(
+                    ColumnArrayView<TYPE_BOOLEAN>::create(left_column));
             break;
         case TYPE_TINYINT:
-            res = _execute_number_expanded<TYPE_TINYINT, TYPE_SMALLINT>(offsets, *nested_column,
-                                                                        nested_null_map);
+            res = _execute_number_expanded<TYPE_TINYINT, TYPE_SMALLINT>(
+                    ColumnArrayView<TYPE_TINYINT>::create(left_column));
             break;
         case TYPE_SMALLINT:
-            res = _execute_number_expanded<TYPE_SMALLINT, TYPE_INT>(offsets, *nested_column,
-                                                                    nested_null_map);
+            res = _execute_number_expanded<TYPE_SMALLINT, TYPE_INT>(
+                    ColumnArrayView<TYPE_SMALLINT>::create(left_column));
             break;
         case TYPE_INT:
-            res = _execute_number_expanded<TYPE_INT, TYPE_BIGINT>(offsets, *nested_column,
-                                                                  nested_null_map);
+            res = _execute_number_expanded<TYPE_INT, TYPE_BIGINT>(
+                    ColumnArrayView<TYPE_INT>::create(left_column));
             break;
         case TYPE_BIGINT:
-            res = _execute_number_expanded<TYPE_BIGINT, TYPE_LARGEINT>(offsets, *nested_column,
-                                                                       nested_null_map);
+            res = _execute_number_expanded<TYPE_BIGINT, TYPE_LARGEINT>(
+                    ColumnArrayView<TYPE_BIGINT>::create(left_column));
             break;
         case TYPE_LARGEINT:
-            res = _execute_number_expanded<TYPE_LARGEINT, TYPE_LARGEINT>(offsets, *nested_column,
-                                                                         nested_null_map);
+            res = _execute_number_expanded<TYPE_LARGEINT, TYPE_LARGEINT>(
+                    ColumnArrayView<TYPE_LARGEINT>::create(left_column));
             break;
         case TYPE_FLOAT:
-            res = _execute_number_expanded<TYPE_FLOAT, TYPE_DOUBLE>(offsets, *nested_column,
-                                                                    nested_null_map);
+            res = _execute_number_expanded<TYPE_FLOAT, TYPE_DOUBLE>(
+                    ColumnArrayView<TYPE_FLOAT>::create(left_column));
             break;
         case TYPE_DOUBLE:
-            res = _execute_number_expanded<TYPE_DOUBLE, TYPE_DOUBLE>(offsets, *nested_column,
-                                                                     nested_null_map);
+            res = _execute_number_expanded<TYPE_DOUBLE, TYPE_DOUBLE>(
+                    ColumnArrayView<TYPE_DOUBLE>::create(left_column));
             break;
         case TYPE_DECIMAL32:
-            res = _execute_number_expanded<TYPE_DECIMAL32, TYPE_DECIMAL32>(offsets, *nested_column,
-                                                                           nested_null_map);
+            res = _execute_number_expanded<TYPE_DECIMAL32, TYPE_DECIMAL32>(
+                    ColumnArrayView<TYPE_DECIMAL32>::create(left_column));
             break;
         case TYPE_DECIMAL64:
-            res = _execute_number_expanded<TYPE_DECIMAL64, TYPE_DECIMAL64>(offsets, *nested_column,
-                                                                           nested_null_map);
+            res = _execute_number_expanded<TYPE_DECIMAL64, TYPE_DECIMAL64>(
+                    ColumnArrayView<TYPE_DECIMAL64>::create(left_column));
             break;
         case TYPE_DECIMAL128I:
             res = _execute_number_expanded<TYPE_DECIMAL128I, TYPE_DECIMAL128I>(
-                    offsets, *nested_column, nested_null_map);
+                    ColumnArrayView<TYPE_DECIMAL128I>::create(left_column));
             break;
         case TYPE_DECIMALV2:
-            res = _execute_number_expanded<TYPE_DECIMALV2, TYPE_DECIMALV2>(offsets, *nested_column,
-                                                                           nested_null_map);
+            res = _execute_number_expanded<TYPE_DECIMALV2, TYPE_DECIMALV2>(
+                    ColumnArrayView<TYPE_DECIMALV2>::create(left_column));
             break;
         case TYPE_DECIMAL256:
             res = _execute_number_expanded<TYPE_DECIMAL256, TYPE_DECIMAL256>(
-                    offsets, *nested_column, nested_null_map);
+                    ColumnArrayView<TYPE_DECIMAL256>::create(left_column));
             break;
         default:
             return nullptr;
diff --git a/be/src/exprs/function/array/function_array_distance.h b/be/src/exprs/function/array/function_array_distance.h
index d601b8c2988e02..ab9f50b41ca629 100644
--- a/be/src/exprs/function/array/function_array_distance.h
+++ b/be/src/exprs/function/array/function_array_distance.h
@@ -138,7 +138,8 @@ class FunctionArrayDistance : public IFunction {
 
         // Check inner nullable (array elements)
         const auto& array_col = assert_cast<const ColumnArray&>(*raw);
-        if (is_column_nullable(*array_col.get_data_ptr()) && array_col.get_data_ptr()->has_null()) {
+        const auto& nested_nullable = assert_cast<const ColumnNullable&>(array_col.get_data());
+        if (nested_nullable.has_null()) {
             throw doris::Exception(ErrorCode::INVALID_ARGUMENT,
                                    "{} for function {} cannot have null", arg_name, func_name);
         }
diff --git a/be/src/exprs/function/array/function_array_element.h b/be/src/exprs/function/array/function_array_element.h
index 9b93680846b91d..f851d14763ad14 100644
--- a/be/src/exprs/function/array/function_array_element.h
+++ b/be/src/exprs/function/array/function_array_element.h
@@ -464,16 +464,10 @@ class FunctionArrayElement : public IFunction {
         const auto& array_column = assert_cast<const ColumnArray&>(*arguments[0].column);
         const auto& offsets = array_column.get_offsets();
         DCHECK(is_const_array ? offsets.size() == 1 : offsets.size() == input_rows_count);
-        const UInt8* nested_null_map = nullptr;
-        ColumnPtr nested_column = nullptr;
-        if (is_column_nullable(array_column.get_data())) {
-            const auto& nested_null_column =
-                    reinterpret_cast<const ColumnNullable&>(array_column.get_data());
-            nested_null_map = nested_null_column.get_null_map_column().get_data().data();
-            nested_column = nested_null_column.get_nested_column_ptr();
-        } else {
-            nested_column = array_column.get_data_ptr();
-        }
+        const auto& nested_null_column =
+                assert_cast<const ColumnNullable&>(array_column.get_data());
+        const UInt8* nested_null_map = nested_null_column.get_null_map_data().data();
+        ColumnPtr nested_column = nested_null_column.get_nested_column_ptr();
 
         ColumnPtr res = nullptr;
         auto left_element_type = remove_nullable(
diff --git a/be/src/exprs/function/array/function_array_enumerate_uniq.cpp b/be/src/exprs/function/array/function_array_enumerate_uniq.cpp
index c91b3a420b3599..80f77fb209b7e4 100644
--- a/be/src/exprs/function/array/function_array_enumerate_uniq.cpp
+++ b/be/src/exprs/function/array/function_array_enumerate_uniq.cpp
@@ -121,8 +121,6 @@ class FunctionArrayEnumerateUniq : public IFunction {
         ColumnPtr src_offsets;
         Columns src_columns; // to keep ownership
 
-        const ColumnArray* first_column_array = nullptr;
-
         for (size_t i = 0; i < arguments.size(); i++) {
             src_columns.emplace_back(
                     block.get_by_position(arguments[i]).column->convert_to_full_column_if_const());
@@ -137,7 +135,6 @@ class FunctionArrayEnumerateUniq : public IFunction {
 
             const ColumnArray::Offsets64& cur_offsets = array->get_offsets();
             if (i == 0) {
-                first_column_array = array;
                 offsets = &cur_offsets;
                 src_offsets = array->get_offsets_ptr();
             } else if (*offsets != cur_offsets) {
@@ -149,11 +146,10 @@ class FunctionArrayEnumerateUniq : public IFunction {
         }
 
         const NullMapType* null_map = nullptr;
-        if (arguments.size() == 1 &&
-            (nullptr != check_and_get_column<ColumnNullable>(data_columns[0]))) {
-            const auto* nullable = check_and_get_column<ColumnNullable>(data_columns[0]);
+        if (arguments.size() == 1) {
+            const auto* nullable = assert_cast<const ColumnNullable*>(data_columns[0]);
             data_columns[0] = nullable->get_nested_column_ptr().get();
-            null_map = &nullable->get_null_map_column().get_data();
+            null_map = &nullable->get_null_map_data();
         }
 
         auto dst_nested_column = ColumnInt64::create();
@@ -188,11 +184,9 @@ class FunctionArrayEnumerateUniq : public IFunction {
                     data_columns, *offsets, nullptr, dst_values);
         }
 
-        ColumnPtr nested_column = dst_nested_column->get_ptr();
-        if (is_column_nullable(first_column_array->get_data())) {
-            nested_column = ColumnNullable::create(nested_column,
-                                                   ColumnUInt8::create(nested_column->size(), 0));
-        }
+        auto dst_null_map = ColumnUInt8::create(dst_nested_column->size(), 0);
+        ColumnPtr nested_column =
+                ColumnNullable::create(std::move(dst_nested_column), std::move(dst_null_map));
         ColumnPtr res_column = ColumnArray::create(std::move(nested_column), src_offsets);
         if (arguments.size() == 1) {
             auto left_column =
diff --git a/be/src/exprs/function/array/function_array_exists.cpp b/be/src/exprs/function/array/function_array_exists.cpp
index 3ae69836270d4c..f16c6ab8db1264 100644
--- a/be/src/exprs/function/array/function_array_exists.cpp
+++ b/be/src/exprs/function/array/function_array_exists.cpp
@@ -30,6 +30,7 @@
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column.h"
 #include "core/column/column_array.h"
+#include "core/column/column_array_view.h"
 #include "core/column/column_nullable.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/data_type.h"
@@ -67,19 +68,18 @@ class FunctionArrayExists : public IFunction {
         // 1. get first array column
         const auto first_column =
                 block.get_by_position(arguments[0]).column->convert_to_full_column_if_const();
+        auto array_view = ColumnArrayView<TYPE_BOOLEAN>::create(first_column);
         const ColumnArray& first_col_array = assert_cast<const ColumnArray&>(*first_column);
         const auto& nested_nullable_column =
                 assert_cast<const ColumnNullable&>(*first_col_array.get_data_ptr());
-        const auto nested_column = nested_nullable_column.get_nested_column_ptr();
-        const size_t nested_column_size = nested_column->size();
+        const size_t nested_column_size = array_view.element_data.size();
         ColumnPtr result_null_map = nested_nullable_column.get_null_map_column_ptr();
 
         // 2. compute result
         auto result_column = ColumnUInt8::create(nested_column_size, 0);
         auto* __restrict result_column_data = result_column->get_data().data();
         ColumnPtr result_offset_column = first_col_array.get_offsets_ptr();
-        const auto* __restrict nested_column_data =
-                assert_cast<const ColumnUInt8&>(*nested_column).get_data().data();
+        const auto* __restrict nested_column_data = array_view.get_data();
 
         for (size_t row = 0; row < nested_column_size; ++row) {
             result_column_data[row] = nested_column_data[row] != 0;
diff --git a/be/src/exprs/function/array/function_array_filter.cpp b/be/src/exprs/function/array/function_array_filter.cpp
index 9d9329fa010560..fbcd6f20b57730 100644
--- a/be/src/exprs/function/array/function_array_filter.cpp
+++ b/be/src/exprs/function/array/function_array_filter.cpp
@@ -28,6 +28,7 @@
 #include "core/block/column_numbers.h"
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column.h"
+#include "core/column/column_array_view.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/data_type.h"
 #include "core/types.h"
@@ -66,26 +67,14 @@ class FunctionArrayFilter : public IFunction {
         //TODO: maybe need optimize not convert
         auto first_column =
                 block.get_by_position(arguments[0]).column->convert_to_full_column_if_const();
-        auto second_column =
-                block.get_by_position(arguments[1]).column->convert_to_full_column_if_const();
+        auto second_array_view =
+                ColumnArrayView<TYPE_BOOLEAN>::create(block.get_by_position(arguments[1]).column);
 
         const ColumnArray& first_col_array = assert_cast<const ColumnArray&>(*first_column);
         const auto& first_off_data = first_col_array.get_offsets_column().get_data();
         const auto& first_nested_nullable_column =
                 assert_cast<const ColumnNullable&>(*first_col_array.get_data_ptr());
 
-        const ColumnArray& second_col_array = assert_cast<const ColumnArray&>(*second_column);
-        const auto& second_off_data = second_col_array.get_offsets_column().get_data();
-        const auto& second_nested_null_map_data =
-                assert_cast<const ColumnNullable&>(*second_col_array.get_data_ptr())
-                        .get_null_map_column()
-                        .get_data();
-        const auto& second_nested_column =
-                assert_cast<const ColumnNullable&>(*second_col_array.get_data_ptr())
-                        .get_nested_column();
-        const auto& second_nested_data =
-                assert_cast<const ColumnUInt8&>(second_nested_column).get_data();
-
         auto result_data_column = first_nested_nullable_column.clone_empty();
         auto result_offset_column = ColumnArray::ColumnOffsets::create();
         auto& result_offset_data = result_offset_column->get_data();
@@ -97,18 +86,18 @@ class FunctionArrayFilter : public IFunction {
             unsigned long count = 0;
             auto first_offset_start = first_off_data[row - 1];
             auto first_offset_end = first_off_data[row];
-            auto second_offset_start = second_off_data[row - 1];
-            auto second_offset_end = second_off_data[row];
-            auto move_off = second_offset_start;
+            auto filter_data = second_array_view[row];
+            const auto* filter_values = filter_data.get_data();
+            const auto* filter_null_map = filter_data.get_null_map_data();
+            size_t filter_pos = 0;
             for (auto off = first_offset_start;
-                 off < first_offset_end && move_off < second_offset_end; // not out range
-                 ++off) {
-                if (second_nested_null_map_data[move_off] == 0 && // not null
-                    second_nested_data[move_off] == 1) {          // not 0
+                 off < first_offset_end && filter_pos < filter_data.size(); // not out range
+                 ++off, ++filter_pos) {
+                if (!filter_null_map[filter_pos] &&   // not null
+                    filter_values[filter_pos] == 1) { // not 0
                     count++;
                     selector.push_back(off);
                 }
-                move_off++;
             }
             result_offset_data.push_back(count + result_offset_data.back());
         }
diff --git a/be/src/exprs/function/array/varray_match_function.cpp b/be/src/exprs/function/array/varray_match_function.cpp
index fed2c2d20b7685..5a31590eec47df 100644
--- a/be/src/exprs/function/array/varray_match_function.cpp
+++ b/be/src/exprs/function/array/varray_match_function.cpp
@@ -20,12 +20,10 @@
 #include <utility>
 
 #include "common/status.h"
-#include "core/assert_cast.h"
 #include "core/block/block.h"
 #include "core/block/column_numbers.h"
 #include "core/block/column_with_type_and_name.h"
-#include "core/column/column.h"
-#include "core/column/column_array.h"
+#include "core/column/column_array_view.h"
 #include "core/column/column_nullable.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/data_type_number.h" // IWYU pragma: keep
@@ -58,39 +56,8 @@ class ArrayMatchFunction : public IFunction {
     Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments,
                         uint32_t result, size_t input_rows_count) const override {
         // here is executed by array_map filtered and arg[0] is bool result column
-        const auto& [src_column, src_const] =
-                unpack_if_const(block.get_by_position(arguments[0]).column);
-        const ColumnArray* array_column = nullptr;
-        const UInt8* array_null_map = nullptr;
-        if (const auto* nullable_array = check_and_get_column<ColumnNullable>(src_column.get())) {
-            array_column = assert_cast<const ColumnArray*>(&nullable_array->get_nested_column());
-            array_null_map = nullable_array->get_null_map_column().get_data().data();
-        } else {
-            array_column = assert_cast<const ColumnArray*>(src_column.get());
-        }
-
-        if (!array_column) {
-            return Status::RuntimeError("unsupported types for function {}({})", get_name(),
-                                        block.get_by_position(arguments[0]).type->get_name());
-        }
-
-        const auto& offsets = array_column->get_offsets();
-        ColumnPtr nested_column = nullptr;
-        const UInt8* nested_null_map = nullptr;
-        if (const auto* nested_null_column =
-                    check_and_get_column<ColumnNullable>(&array_column->get_data())) {
-            nested_null_map = nested_null_column->get_null_map_column().get_data().data();
-            nested_column = nested_null_column->get_nested_column_ptr();
-        } else {
-            nested_column = array_column->get_data_ptr();
-        }
-
-        if (!nested_column) {
-            return Status::RuntimeError("unsupported types for function {}({})", get_name(),
-                                        block.get_by_position(arguments[0]).type->get_name());
-        }
-
-        const auto& nested_data = assert_cast<const ColumnUInt8&>(*nested_column).get_data();
+        auto array_view =
+                ColumnArrayView<TYPE_BOOLEAN>::create(block.get_by_position(arguments[0]).column);
 
         // result is nullable bool column for every array column
         auto result_data_column = ColumnUInt8::create(input_rows_count, 1);
@@ -98,7 +65,7 @@ class ArrayMatchFunction : public IFunction {
 
         // iterate over all arrays with bool elements
         for (int row = 0; row < input_rows_count; ++row) {
-            if (array_null_map && array_null_map[row]) {
+            if (array_view.is_null_at(row)) {
                 // current array is null, this is always null
                 result_null_column->get_data()[row] = 1;
                 result_data_column->get_data()[row] = 0;
@@ -108,11 +75,14 @@ class ArrayMatchFunction : public IFunction {
                 bool has_null_elem = false;
                 // res for current array
                 bool res_for_array = MATCH_ALL;
-                for (auto off = offsets[row - 1]; off < offsets[row]; ++off) {
-                    if (nested_null_map && nested_null_map[off]) {
+                auto array_data = array_view[row];
+                const auto* data = array_data.get_data();
+                const auto* null_map = array_data.get_null_map_data();
+                for (size_t pos = 0; pos < array_data.size(); ++pos) {
+                    if (null_map[pos]) {
                         has_null_elem = true;
                     } else {
-                        if (nested_data[off] != MATCH_ALL) { // not match
+                        if (data[pos] != MATCH_ALL) { // not match
                             res_for_array = !MATCH_ALL;
                             break;
                         } // default is MATCH_ALL
diff --git a/be/src/exprs/lambda_function/varray_filter_function.cpp b/be/src/exprs/lambda_function/varray_filter_function.cpp
index e4bb3897c1b26d..b107133a2c8712 100644
--- a/be/src/exprs/lambda_function/varray_filter_function.cpp
+++ b/be/src/exprs/lambda_function/varray_filter_function.cpp
@@ -28,6 +28,7 @@
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column.h"
 #include "core/column/column_array.h"
+#include "core/column/column_array_view.h"
 #include "core/column/column_nullable.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/data_type.h"
@@ -70,7 +71,7 @@ class ArrayFilterFunction : public LambdaFunction {
         //2. get first and second array column
         auto first_column = column_ptr_0->convert_to_full_column_if_const();
 
-        auto second_column = column_ptr_1->convert_to_full_column_if_const();
+        auto second_array_view = ColumnArrayView<TYPE_BOOLEAN>::create(column_ptr_1);
 
         auto input_rows = first_column->size();
         auto first_outside_null_map = ColumnUInt8::create(input_rows, 0);
@@ -95,50 +96,27 @@ class ArrayFilterFunction : public LambdaFunction {
         selector.reserve(first_off_data.size());
         result_offset_data.reserve(input_rows);
 
-        auto second_arg_column = second_column;
-        auto second_outside_null_map = ColumnUInt8::create(input_rows, 0);
-        if (is_column_nullable(*second_arg_column)) {
-            second_arg_column = assert_cast<const ColumnNullable*>(second_column.get())
-                                        ->get_nested_column_ptr();
-            const auto& column_array_nullmap =
-                    assert_cast<const ColumnNullable*>(second_column.get())->get_null_map_column();
-            VectorizedUtils::update_null_map(second_outside_null_map->get_data(),
-                                             column_array_nullmap.get_data());
-        }
-        const auto& second_col_array = assert_cast<const ColumnArray&>(*second_arg_column);
-        const auto& second_off_data = second_col_array.get_offsets_column().get_data();
-        const auto& second_nested_null_map_data =
-                assert_cast<const ColumnNullable&>(*second_col_array.get_data_ptr())
-                        .get_null_map_column()
-                        .get_data();
-        const auto& second_nested_column =
-                assert_cast<const ColumnNullable&>(*second_col_array.get_data_ptr())
-                        .get_nested_column();
-        const auto& second_nested_data =
-                assert_cast<const ColumnUInt8&>(second_nested_column).get_data();
-
         //3. get the idx of second column data is not null and not 0
         for (int row = 0; row < input_rows; ++row) {
             //first or second column is null, so current row is invalid data
-            if (first_outside_null_map->get_data()[row] ||
-                second_outside_null_map->get_data()[row]) {
+            if (first_outside_null_map->get_data()[row] || second_array_view.is_null_at(row)) {
                 result_offset_data.push_back(result_offset_data.back());
             } else {
                 unsigned long count = 0;
                 auto first_offset_start = first_off_data[row - 1];
                 auto first_offset_end = first_off_data[row];
-                auto second_offset_start = second_off_data[row - 1];
-                auto second_offset_end = second_off_data[row];
-                auto move_off = second_offset_start;
+                auto filter_data = second_array_view[row];
+                const auto* filter_values = filter_data.get_data();
+                const auto* filter_null_map = filter_data.get_null_map_data();
+                size_t filter_pos = 0;
                 for (auto off = first_offset_start;
-                     off < first_offset_end && move_off < second_offset_end; // not out range
-                     ++off) {
-                    if (!second_nested_null_map_data[move_off] && // not null
-                        second_nested_data[move_off]) {           // not 0
+                     off < first_offset_end && filter_pos < filter_data.size(); // not out range
+                     ++off, ++filter_pos) {
+                    if (!filter_null_map[filter_pos] && // not null
+                        filter_values[filter_pos]) {    // not 0
                         count++;
                         selector.push_back(off);
                     }
-                    move_off++;
                 }
                 result_offset_data.push_back(count + result_offset_data.back());
             }
@@ -153,7 +131,7 @@ class ArrayFilterFunction : public LambdaFunction {
                                            std::move(first_outside_null_map));
         } else {
             DCHECK(!first_column->is_nullable());
-            DCHECK(!second_column->is_nullable());
+            DCHECK(!second_array_view.is_nullable());
             result_column = ColumnArray::create(std::move(result_data_column),
                                                 std::move(result_offset_column));
         }
diff --git a/be/test/exprs/function/function_array_element_test.cpp b/be/test/exprs/function/function_array_element_test.cpp
index 7d3c9bceccf364..880bc7cb246b3f 100644
--- a/be/test/exprs/function/function_array_element_test.cpp
+++ b/be/test/exprs/function/function_array_element_test.cpp
@@ -148,7 +148,9 @@ static ColumnPtr make_const_int32_array(std::vector<Int32> values, size_t appare
     }
     auto offsets = ColumnArray::ColumnOffsets::create();
     offsets->insert_value(static_cast<Int64>(values.size()));
-    auto arr = ColumnArray::create(std::move(data_col), std::move(offsets));
+    auto nested_null_map = ColumnUInt8::create(values.size(), 0);
+    auto nested_nullable = ColumnNullable::create(std::move(data_col), std::move(nested_null_map));
+    auto arr = ColumnArray::create(std::move(nested_nullable), std::move(offsets));
     // element_at always produces Nullable, so use a non-null wrapper
     auto null_map = ColumnUInt8::create(1, 0 /*not null*/);
     auto nullable_arr = ColumnNullable::create(std::move(arr), std::move(null_map));
@@ -163,7 +165,9 @@ static ColumnPtr make_const_string_array(std::vector<std::string> values, size_t
     }
     auto offsets = ColumnArray::ColumnOffsets::create();
     offsets->insert_value(static_cast<Int64>(values.size()));
-    auto arr = ColumnArray::create(std::move(data_col), std::move(offsets));
+    auto nested_null_map = ColumnUInt8::create(values.size(), 0);
+    auto nested_nullable = ColumnNullable::create(std::move(data_col), std::move(nested_null_map));
+    auto arr = ColumnArray::create(std::move(nested_nullable), std::move(offsets));
     auto null_map = ColumnUInt8::create(1, 0);
     auto nullable_arr = ColumnNullable::create(std::move(arr), std::move(null_map));
     return ColumnConst::create(std::move(nullable_arr), apparent_size);
@@ -240,7 +244,10 @@ TEST(function_array_element_test, element_at_const_null_array) {
     auto inner_data = ColumnInt32::create();
     auto inner_offsets = ColumnArray::ColumnOffsets::create();
     inner_offsets->insert_value(0);
-    auto inner_arr = ColumnArray::create(std::move(inner_data), std::move(inner_offsets));
+    auto inner_nested_nullable =
+            ColumnNullable::create(std::move(inner_data), ColumnUInt8::create());
+    auto inner_arr =
+            ColumnArray::create(std::move(inner_nested_nullable), std::move(inner_offsets));
     auto null_map = ColumnUInt8::create(1, 1 /*null*/);
     auto nullable_arr = ColumnNullable::create(std::move(inner_arr), std::move(null_map));
     ColumnPtr const_arr = ColumnConst::create(std::move(nullable_arr), N);