diff --git a/be/src/exprs/aggregate/aggregate_function_reader_first_last.h b/be/src/exprs/aggregate/aggregate_function_reader_first_last.h
index 33fc9d5574972f..4a2caaee43feaf 100644
--- a/be/src/exprs/aggregate/aggregate_function_reader_first_last.h
+++ b/be/src/exprs/aggregate/aggregate_function_reader_first_last.h
@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include <type_traits>
+
 #include "core/column/column_array.h"
 #include "core/column/column_map.h"
 #include "core/column/column_nullable.h"
@@ -32,7 +34,7 @@
 namespace doris {
 #include "common/compile_check_begin.h"
 
-template <typename ColVecType, bool arg_is_nullable>
+template <bool arg_is_nullable>
 struct Value {
 public:
     bool is_null() const {
@@ -46,6 +48,7 @@ struct Value {
         return false;
     }
 
+    template <typename ColVecType>
     void insert_into(IColumn& to) const {
         if constexpr (arg_is_nullable) {
             const auto* col = assert_cast<const ColumnNullable*, TypeCheckOnRelease::DISABLE>(_ptr);
@@ -56,6 +59,17 @@ struct Value {
         }
     }
 
+    // Non-template version: virtual dispatch on IColumn::insert_from.
+    // Used by window path where devirtualization is not beneficial.
+    void insert_into(IColumn& to) const {
+        if constexpr (arg_is_nullable) {
+            const auto* col = assert_cast<const ColumnNullable*, TypeCheckOnRelease::DISABLE>(_ptr);
+            to.insert_from(col->get_nested_column(), _offset);
+        } else {
+            to.insert_from(*_ptr, _offset);
+        }
+    }
+
     void set_value(const IColumn* column, size_t row) {
         _ptr = column;
         _offset = row;
@@ -71,15 +85,17 @@ struct Value {
     size_t _offset = 0;
 };
 
-template <typename ColVecType, bool arg_is_nullable>
-struct CopiedValue : public Value<ColVecType, arg_is_nullable> {
+template <bool arg_is_nullable>
+struct CopiedValue : public Value<arg_is_nullable> {
 public:
+    template <typename ColVecType>
     void insert_into(IColumn& to) const {
         assert_cast<ColVecType&, TypeCheckOnRelease::DISABLE>(to).insert(_copied_value);
     }
 
     bool is_null() const { return this->_ptr == nullptr; }
 
+    template <typename ColVecType>
     void set_value(const IColumn* column, size_t row) {
         // here _ptr, maybe null at row, so call reset to set nullptr
         // But we will use is_null() check first, others have set _ptr column to a meaningless address
@@ -108,8 +124,8 @@ struct CopiedValue : public Value<ColVecType, arg_is_nullable> {
 template <typename ColVecType, bool result_is_nullable, bool arg_is_nullable, bool is_copy>
 struct ReaderFirstAndLastData {
 public:
-    using StoreType = std::conditional_t<is_copy, CopiedValue<ColVecType, arg_is_nullable>,
-                                         Value<ColVecType, arg_is_nullable>>;
+    using StoreType =
+            std::conditional_t<is_copy, CopiedValue<arg_is_nullable>, Value<arg_is_nullable>>;
     static constexpr bool nullable = arg_is_nullable;
     static constexpr bool result_nullable = result_is_nullable;
 
@@ -126,17 +142,29 @@ struct ReaderFirstAndLastData {
             } else {
                 auto& col = assert_cast<ColumnNullable&>(to);
                 col.get_null_map_data().push_back(0);
-                _data_value.insert_into(col.get_nested_column());
+                if constexpr (!std::is_same_v<ColVecType, void>) {
+                    _data_value.template insert_into<ColVecType>(col.get_nested_column());
+                } else {
+                    _data_value.insert_into(col.get_nested_column());
+                }
             }
         } else {
-            _data_value.insert_into(to);
+            if constexpr (!std::is_same_v<ColVecType, void>) {
+                _data_value.template insert_into<ColVecType>(to);
+            } else {
+                _data_value.insert_into(to);
+            }
         }
     }
 
     // here not check the columns[0] is null at the row,
     // but it is need to check in other
     void set_value(const IColumn** columns, size_t pos) {
-        _data_value.set_value(columns[0], pos);
+        if constexpr (is_copy) {
+            _data_value.template set_value<ColVecType>(columns[0], pos);
+        } else {
+            _data_value.set_value(columns[0], pos);
+        }
         _has_value = true;
     }
 
diff --git a/be/src/exprs/aggregate/aggregate_function_window.cpp b/be/src/exprs/aggregate/aggregate_function_window.cpp
index 01e35bb999a3e1..41bbebdf0c8f50 100644
--- a/be/src/exprs/aggregate/aggregate_function_window.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_window.cpp
@@ -49,11 +49,17 @@ AggregateFunctionPtr create_aggregate_function_window_first(const std::string& n
                                                             const DataTypePtr& result_type,
                                                             const bool result_is_nullable,
                                                             const AggregateFunctionAttr& attr);
+AggregateFunctionPtr create_aggregate_function_window_first_ignore_null(
+        const std::string& name, const DataTypes& argument_types, const DataTypePtr& result_type,
+        const bool result_is_nullable, const AggregateFunctionAttr& attr);
 AggregateFunctionPtr create_aggregate_function_window_last(const std::string& name,
                                                            const DataTypes& argument_types,
                                                            const DataTypePtr& result_type,
                                                            const bool result_is_nullable,
                                                            const AggregateFunctionAttr& attr);
+AggregateFunctionPtr create_aggregate_function_window_last_ignore_null(
+        const std::string& name, const DataTypes& argument_types, const DataTypePtr& result_type,
+        const bool result_is_nullable, const AggregateFunctionAttr& attr);
 AggregateFunctionPtr create_aggregate_function_window_nth_value(const std::string& name,
                                                                 const DataTypes& argument_types,
                                                                 const DataTypePtr& result_type,
@@ -89,8 +95,36 @@ void register_aggregate_function_window_lead_lag_first_last(
         AggregateFunctionSimpleFactory& factory) {
     factory.register_function_both("lead", create_aggregate_function_window_lead);
     factory.register_function_both("lag", create_aggregate_function_window_lag);
-    factory.register_function_both("first_value", create_aggregate_function_window_first);
-    factory.register_function_both("last_value", create_aggregate_function_window_last);
+    // FE rewrites first_value(k1, false) → first_value(k1), so argument_types.size() == 2
+    // means arg_ignore_null = true. Dispatch at registration to avoid runtime branching
+    // that would double template instantiations.
+    factory.register_function_both(
+            "first_value",
+            [](const std::string& name, const DataTypes& argument_types,
+               const DataTypePtr& result_type, const bool result_is_nullable,
+               const AggregateFunctionAttr& attr) -> AggregateFunctionPtr {
+                if (argument_types.size() == 2) {
+                    return create_aggregate_function_window_first_ignore_null(
+                            name, argument_types, result_type, result_is_nullable, attr);
+                }
+                return create_aggregate_function_window_first(name, argument_types, result_type,
+                                                              result_is_nullable, attr);
+            });
+    factory.register_function_both(
+            "last_value",
+            [](const std::string& name, const DataTypes& argument_types,
+               const DataTypePtr& result_type, const bool result_is_nullable,
+               const AggregateFunctionAttr& attr) -> AggregateFunctionPtr {
+                if (argument_types.size() == 2) {
+                    return create_aggregate_function_window_last_ignore_null(
+                            name, argument_types, result_type, result_is_nullable, attr);
+                }
+                return create_aggregate_function_window_last(name, argument_types, result_type,
+                                                             result_is_nullable, attr);
+            });
+    // nth_value always has 2 args (column, N) from FE.
+    // WindowFunctionNthValueImpl does not implement ignore-null logic,
+    // so register directly without dispatch.
     factory.register_function_both("nth_value", create_aggregate_function_window_nth_value);
 }
 
diff --git a/be/src/exprs/aggregate/aggregate_function_window.h b/be/src/exprs/aggregate/aggregate_function_window.h
index fe7556c47b6738..9b654c77ff5eb1 100644
--- a/be/src/exprs/aggregate/aggregate_function_window.h
+++ b/be/src/exprs/aggregate/aggregate_function_window.h
@@ -397,15 +397,15 @@ class WindowFunctionNTile final
     void deserialize(AggregateDataPtr place, BufferReadable& buf, Arena&) const override {}
 };
 
-template <typename ColVecType, bool result_is_nullable, bool arg_is_nullable>
+template <bool result_is_nullable, bool arg_is_nullable>
 struct FirstLastData
-        : public ReaderFirstAndLastData<ColVecType, result_is_nullable, arg_is_nullable, false> {
+        : public ReaderFirstAndLastData<void, result_is_nullable, arg_is_nullable, false> {
 public:
     void set_is_null() { this->_data_value.reset(); }
 };
 
-template <typename ColVecType, bool result_is_nullable, bool arg_is_nullable>
-struct NthValueData : public FirstLastData<ColVecType, result_is_nullable, arg_is_nullable> {
+template <bool result_is_nullable, bool arg_is_nullable>
+struct NthValueData : public FirstLastData<result_is_nullable, arg_is_nullable> {
 public:
     void reset() {
         this->_data_value.reset();
@@ -418,8 +418,8 @@ struct NthValueData : public FirstLastData<ColVecType, result_is_nullable, arg_i
     int64_t _frame_total_rows = 0;
 };
 
-template <typename ColVecType, bool arg_is_nullable>
-struct BaseValue : public Value<ColVecType, arg_is_nullable> {
+template <bool arg_is_nullable>
+struct BaseValue : public Value<arg_is_nullable> {
 public:
     bool is_null() const { return this->_ptr == nullptr; }
     // because _ptr pointer to first_argument or third argument, so it's difficult to cast ptr
@@ -427,7 +427,7 @@ struct BaseValue : public Value<ColVecType, arg_is_nullable> {
     StringRef get_value() const { return this->_ptr->get_data_at(this->_offset); }
 };
 
-template <typename ColVecType, bool result_is_nullable, bool arg_is_nullable>
+template <bool result_is_nullable, bool arg_is_nullable>
 struct LeadLagData {
 public:
     static constexpr bool result_nullable = result_is_nullable;
@@ -492,7 +492,7 @@ struct LeadLagData {
     int64_t get_offset_value() const { return _offset_value; }
 
 private:
-    BaseValue<ColVecType, arg_is_nullable> _data_value;
+    BaseValue<arg_is_nullable> _data_value;
     bool _is_inited = false;
     int64_t _offset_value = 0;
 };
diff --git a/be/src/exprs/aggregate/aggregate_function_window_first.cpp b/be/src/exprs/aggregate/aggregate_function_window_first.cpp
index 63b7054d5d40d7..79051c09665dc8 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_first.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_window_first.cpp
@@ -20,8 +20,11 @@
 namespace doris {
 #include "common/compile_check_begin.h"
 
-CREATE_WINDOW_FUNCTION_WITH_NAME_AND_DATA(create_aggregate_function_window_first, FirstLastData,
-                                          WindowFunctionFirstImpl);
+CREATE_WINDOW_FUNCTION_DIRECT(create_aggregate_function_window_first, FirstLastData,
+                              WindowFunctionFirstImpl, false);
+
+CREATE_WINDOW_FUNCTION_DIRECT(create_aggregate_function_window_first_ignore_null, FirstLastData,
+                              WindowFunctionFirstImpl, true);
 
 #include "common/compile_check_end.h"
 } // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_window_impl.h b/be/src/exprs/aggregate/aggregate_function_window_impl.h
index c3f6f0d4033156..09fbb199736779 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_impl.h
+++ b/be/src/exprs/aggregate/aggregate_function_window_impl.h
@@ -29,328 +29,30 @@
 namespace doris {
 #include "common/compile_check_begin.h"
 
-template <template <typename> class AggregateFunctionTemplate,
-          template <typename ColVecType, bool, bool> class Data,
-          template <typename, bool> class Impl, bool result_is_nullable, bool arg_is_nullable>
-AggregateFunctionPtr create_function_lead_lag_first_last(const String& name,
-                                                         const DataTypes& argument_types) {
-    bool arg_ignore_null_value = false;
-    // FE have rewrite case first_value(k1,false)--->first_value(k1)
-    // so size is 2, must will be arg_ignore_null_value
-    if (argument_types.size() == 2) {
-        DCHECK(name == "first_value" || name == "last_value" || name == "nth_value")
-                << "invalid function name: " << name;
-        arg_ignore_null_value = true;
-    }
-    switch (argument_types[0]->get_primitive_type()) {
-    case PrimitiveType::TYPE_BOOLEAN: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnUInt8, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnUInt8, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_TINYINT: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt8, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt8, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_SMALLINT: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt16, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt16, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_INT: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt32, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt32, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_BIGINT: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt64, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt64, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_LARGEINT: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt128, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnInt128, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_FLOAT: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnFloat32, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnFloat32, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_DOUBLE: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnFloat64, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnFloat64, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_DECIMAL32: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal32, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal32, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_DECIMAL64: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal64, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal64, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_DECIMAL128I: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal128V3, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal128V3, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_DECIMALV2: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal128V2, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal128V2, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_DECIMAL256: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal256, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDecimal256, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_STRING:
-    case PrimitiveType::TYPE_CHAR:
-    case PrimitiveType::TYPE_VARCHAR:
-    case PrimitiveType::TYPE_JSONB: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnString, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnString, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_DATETIMEV2: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDateTimeV2, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDateTimeV2, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_DATEV2: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDateV2, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnDateV2, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_IPV4: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnIPv4, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnIPv4, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_IPV6: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnIPv6, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnIPv6, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_ARRAY: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnArray, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnArray, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_MAP: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnMap, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnMap, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_STRUCT: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnStruct, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnStruct, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_VARIANT: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnVariant, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnVariant, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_BITMAP: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnBitmap, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnBitmap, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_HLL: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnHLL, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnHLL, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    case PrimitiveType::TYPE_QUANTILE_STATE: {
-        if (arg_ignore_null_value) {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnQuantileState, result_is_nullable, arg_is_nullable>, true>>>(
-                    argument_types);
-        } else {
-            return std::make_shared<AggregateFunctionTemplate<
-                    Impl<Data<ColumnQuantileState, result_is_nullable, arg_is_nullable>, false>>>(
-                    argument_types);
-        }
-    }
-    default:
-        LOG(WARNING) << "with unknowed type, failed in  create_aggregate_function_" << name
-                     << " and type is: " << argument_types[0]->get_name();
-        return nullptr;
-    }
-}
-
-#define CREATE_WINDOW_FUNCTION_WITH_NAME_AND_DATA(CREATE_FUNCTION_NAME, FUNCTION_DATA,             \
-                                                  FUNCTION_IMPL)                                   \
-    AggregateFunctionPtr CREATE_FUNCTION_NAME(                                                     \
-            const std::string& name, const DataTypes& argument_types,                              \
-            const DataTypePtr& result_type, const bool result_is_nullable,                         \
-            const AggregateFunctionAttr& attr) {                                                   \
-        const bool arg_is_nullable = argument_types[0]->is_nullable();                             \
-        AggregateFunctionPtr res = nullptr;                                                        \
-                                                                                                   \
-        std::visit(                                                                                \
-                [&](auto result_is_nullable, auto arg_is_nullable) {                               \
-                    res = AggregateFunctionPtr(                                                    \
-                            create_function_lead_lag_first_last<WindowFunctionData, FUNCTION_DATA, \
-                                                                FUNCTION_IMPL, result_is_nullable, \
-                                                                arg_is_nullable>(name,             \
-                                                                                 argument_types)); \
-                },                                                                                 \
-                make_bool_variant(result_is_nullable), make_bool_variant(arg_is_nullable));        \
-        if (!res) {                                                                                \
-            LOG(WARNING) << " failed in  create_aggregate_function_" << name                       \
-                         << " and type is: " << argument_types[0]->get_name();                     \
-        }                                                                                          \
-        return res;                                                                                \
+// All window-path Data types (LeadLagData, FirstLastData, NthValueData) no longer have
+// ColVecType template parameter. Direct instantiation with only (result_is_nullable, arg_is_nullable).
+// ARG_IGNORE_NULL is a compile-time bool for first_value/last_value ignore-null variants.
+#define CREATE_WINDOW_FUNCTION_DIRECT(CREATE_FUNCTION_NAME, FUNCTION_DATA, FUNCTION_IMPL,     \
+                                      ARG_IGNORE_NULL)                                        \
+    AggregateFunctionPtr CREATE_FUNCTION_NAME(                                                \
+            const std::string& name, const DataTypes& argument_types,                         \
+            const DataTypePtr& result_type, const bool result_is_nullable,                    \
+            const AggregateFunctionAttr& attr) {                                              \
+        const bool arg_is_nullable = argument_types[0]->is_nullable();                        \
+        AggregateFunctionPtr res = nullptr;                                                   \
+                                                                                              \
+        std::visit(                                                                           \
+                [&](auto result_is_nullable, auto arg_is_nullable) {                          \
+                    res = std::make_shared<WindowFunctionData<                                \
+                            FUNCTION_IMPL<FUNCTION_DATA<result_is_nullable, arg_is_nullable>, \
+                                          ARG_IGNORE_NULL>>>(argument_types);                 \
+                },                                                                            \
+                make_bool_variant(result_is_nullable), make_bool_variant(arg_is_nullable));   \
+        if (!res) {                                                                           \
+            LOG(WARNING) << " failed in  create_aggregate_function_" << name                  \
+                         << " and type is: " << argument_types[0]->get_name();                \
+        }                                                                                     \
+        return res;                                                                           \
     }
 
 #include "common/compile_check_end.h"
diff --git a/be/src/exprs/aggregate/aggregate_function_window_lag.cpp b/be/src/exprs/aggregate/aggregate_function_window_lag.cpp
index 73b7af57cfa687..ab7d0c2ab16df5 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_lag.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_window_lag.cpp
@@ -20,8 +20,8 @@
 namespace doris {
 #include "common/compile_check_begin.h"
 
-CREATE_WINDOW_FUNCTION_WITH_NAME_AND_DATA(create_aggregate_function_window_lag, LeadLagData,
-                                          WindowFunctionLagImpl);
+CREATE_WINDOW_FUNCTION_DIRECT(create_aggregate_function_window_lag, LeadLagData,
+                              WindowFunctionLagImpl, false);
 
 #include "common/compile_check_end.h"
 } // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_window_last.cpp b/be/src/exprs/aggregate/aggregate_function_window_last.cpp
index 7b864404817c26..18bb98d40528ff 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_last.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_window_last.cpp
@@ -20,8 +20,11 @@
 namespace doris {
 #include "common/compile_check_begin.h"
 
-CREATE_WINDOW_FUNCTION_WITH_NAME_AND_DATA(create_aggregate_function_window_last, FirstLastData,
-                                          WindowFunctionLastImpl);
+CREATE_WINDOW_FUNCTION_DIRECT(create_aggregate_function_window_last, FirstLastData,
+                              WindowFunctionLastImpl, false);
+
+CREATE_WINDOW_FUNCTION_DIRECT(create_aggregate_function_window_last_ignore_null, FirstLastData,
+                              WindowFunctionLastImpl, true);
 
 #include "common/compile_check_end.h"
 } // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_window_lead.cpp b/be/src/exprs/aggregate/aggregate_function_window_lead.cpp
index dacf7e51e29f1c..4739acc485c5e3 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_lead.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_window_lead.cpp
@@ -20,8 +20,8 @@
 namespace doris {
 #include "common/compile_check_begin.h"
 
-CREATE_WINDOW_FUNCTION_WITH_NAME_AND_DATA(create_aggregate_function_window_lead, LeadLagData,
-                                          WindowFunctionLeadImpl);
+CREATE_WINDOW_FUNCTION_DIRECT(create_aggregate_function_window_lead, LeadLagData,
+                              WindowFunctionLeadImpl, false);
 
 #include "common/compile_check_end.h"
 } // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_window_nth_value.cpp b/be/src/exprs/aggregate/aggregate_function_window_nth_value.cpp
index d2efa2144fb24e..9ce84cfb76a88f 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_nth_value.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_window_nth_value.cpp
@@ -20,8 +20,11 @@
 namespace doris {
 #include "common/compile_check_begin.h"
 
-CREATE_WINDOW_FUNCTION_WITH_NAME_AND_DATA(create_aggregate_function_window_nth_value, NthValueData,
-                                          WindowFunctionNthValueImpl);
+// nth_value always has 2 args (column, N) from FE.
+// WindowFunctionNthValueImpl does not implement ignore-null logic,
+// so always use false.
+CREATE_WINDOW_FUNCTION_DIRECT(create_aggregate_function_window_nth_value, NthValueData,
+                              WindowFunctionNthValueImpl, false);
 
 #include "common/compile_check_end.h"
 } // namespace doris
diff --git a/be/test/exprs/aggregate/agg_replace_test.cpp b/be/test/exprs/aggregate/agg_replace_test.cpp
index 0b082e8c65848f..a0e90ede1cf8e4 100644
--- a/be/test/exprs/aggregate/agg_replace_test.cpp
+++ b/be/test/exprs/aggregate/agg_replace_test.cpp
@@ -313,7 +313,7 @@ class VAggReplaceTest : public testing::Test {
         EXPECT_EQ(input_nums, data_column->size());
         //test Value
         {
-            Value<ColumnType, nullable> value;
+            Value<nullable> value;
             EXPECT_TRUE(value.is_null());
             for (int64_t i = 0; i < input_nums; ++i) {
                 value.set_value(data_column.get(), i);
@@ -332,17 +332,17 @@ class VAggReplaceTest : public testing::Test {
         }
         //test CopiedValue
         {
-            CopiedValue<ColumnType, nullable> value;
+            CopiedValue<nullable> value;
             EXPECT_TRUE(value.is_null());
             for (int64_t i = 0; i < input_nums; ++i) {
-                value.set_value(data_column.get(), i);
+                value.template set_value<ColumnType>(data_column.get(), i);
                 EXPECT_FALSE(value.is_null());
                 auto to_column = data_type->create_column();
                 if constexpr (nullable) {
                     auto& nullable_col = assert_cast<ColumnNullable&>(*to_column);
-                    value.insert_into(nullable_col.get_nested_column());
+                    value.template insert_into<ColumnType>(nullable_col.get_nested_column());
                 } else {
-                    value.insert_into(*to_column);
+                    value.template insert_into<ColumnType>(*to_column);
                 }
                 EXPECT_EQ(1, to_column->size());
                 check_column_basic<DataType, nullable>(to_column.get(), i);
@@ -360,7 +360,7 @@ class VAggReplaceTest : public testing::Test {
         EXPECT_EQ(input_nums + 1, data_column->size());
         //test Value
         {
-            Value<ColumnArray, nullable> value;
+            Value<nullable> value;
             EXPECT_TRUE(value.is_null());
             for (int64_t i = 0; i <= input_nums; ++i) {
                 value.set_value(data_column.get(), i);
@@ -378,17 +378,17 @@ class VAggReplaceTest : public testing::Test {
         }
         //test CopiedValue
         {
-            CopiedValue<ColumnArray, nullable> value;
+            CopiedValue<nullable> value;
             EXPECT_TRUE(value.is_null());
             for (int64_t i = 0; i <= input_nums; ++i) {
-                value.set_value(data_column.get(), i);
+                value.template set_value<ColumnArray>(data_column.get(), i);
                 EXPECT_FALSE(value.is_null());
                 auto to_column = data_type->create_column();
                 if constexpr (nullable) {
                     auto& nullable_col = assert_cast<ColumnNullable&>(*to_column);
-                    value.insert_into(nullable_col.get_nested_column());
+                    value.template insert_into<ColumnArray>(nullable_col.get_nested_column());
                 } else {
-                    value.insert_into(*to_column);
+                    value.template insert_into<ColumnArray>(*to_column);
                 }
                 EXPECT_EQ(1, to_column->size());
                 check_column_array<DataType, nullable>(to_column.get(), i);