diff --git a/be/src/util/simd/bits.h b/be/src/util/simd/bits.h
index 54e2a0cb2c..8edbe72f4a 100644
--- a/be/src/util/simd/bits.h
+++ b/be/src/util/simd/bits.h
@@ -18,6 +18,8 @@
 #pragma once
 
 #include <cstdint>
+#include <cstring>
+#include <vector>
 
 #ifdef __AVX2__
 #include <immintrin.h>
@@ -97,5 +99,26 @@ inline size_t count_zero_num(const int8_t* __restrict data, const uint8_t* __res
     return num;
 }
 
+// TODO: compare with different SIMD implements
+template <class T>
+inline static size_t find_byte(const std::vector<T>& vec, size_t start, T byte) {
+    if (start >= vec.size()) {
+        return start;
+    }
+    const void* p = std::memchr((const void*)(vec.data() + start), byte, vec.size() - start);
+    if (p == nullptr) {
+        return vec.size();
+    }
+    return (T*)p - vec.data();
+}
+
+inline size_t find_nonzero(const std::vector<uint8_t>& vec, size_t start) {
+    return find_byte<uint8_t>(vec, start, 1);
+}
+
+inline size_t find_zero(const std::vector<uint8_t>& vec, size_t start) {
+    return find_byte<uint8_t>(vec, start, 0);
+}
+
 } // namespace simd
 } // namespace doris
diff --git a/be/src/vec/columns/column.cpp b/be/src/vec/columns/column.cpp
index 815be82024..845600bc1b 100644
--- a/be/src/vec/columns/column.cpp
+++ b/be/src/vec/columns/column.cpp
@@ -25,6 +25,7 @@
 #include "vec/columns/column_const.h"
 #include "vec/columns/column_nullable.h"
 #include "vec/core/field.h"
+#include "vec/core/sort_block.h"
 
 namespace doris::vectorized {
 
@@ -46,6 +47,11 @@ void IColumn::insert_from(const IColumn& src, size_t n) {
     insert(src[n]);
 }
 
+void IColumn::sort_column(const ColumnSorter* sorter, EqualFlags& flags,
+                          IColumn::Permutation& perms, EqualRange& range, bool last_column) const {
+    sorter->sort_column(static_cast<const IColumn&>(*this), flags, perms, range, last_column);
+}
+
 bool is_column_nullable(const IColumn& column) {
     return check_column<ColumnNullable>(column);
 }
diff --git a/be/src/vec/columns/column.h b/be/src/vec/columns/column.h
index a1883a1935..71d737274c 100644
--- a/be/src/vec/columns/column.h
+++ b/be/src/vec/columns/column.h
@@ -48,6 +48,10 @@ namespace doris::vectorized {
 class Arena;
 class Field;
 
+class ColumnSorter;
+using EqualFlags = std::vector<uint8_t>;
+using EqualRange = std::pair<int, int>;
+
 /// Declares interface to store columns in memory.
 class IColumn : public COW<IColumn> {
 private:
@@ -529,6 +533,10 @@ public:
 
     virtual bool low_cardinality() const { return false; }
 
+    virtual void sort_column(const ColumnSorter* sorter, EqualFlags& flags,
+                             IColumn::Permutation& perms, EqualRange& range,
+                             bool last_column) const;
+
     virtual ~IColumn() = default;
     IColumn() = default;
     IColumn(const IColumn&) = default;
diff --git a/be/src/vec/columns/column_decimal.cpp b/be/src/vec/columns/column_decimal.cpp
index 91145e4473..ea904c8c30 100644
--- a/be/src/vec/columns/column_decimal.cpp
+++ b/be/src/vec/columns/column_decimal.cpp
@@ -27,6 +27,7 @@
 #include "vec/common/exception.h"
 #include "vec/common/sip_hash.h"
 #include "vec/common/unaligned.h"
+#include "vec/core/sort_block.h"
 
 template <typename T>
 bool decimal_less(T x, T y, doris::vectorized::UInt32 x_scale, doris::vectorized::UInt32 y_scale);
@@ -330,6 +331,13 @@ void ColumnDecimal<T>::get_extremes(Field& min, Field& max) const {
     max = NearestFieldType<T>(cur_max, scale);
 }
 
+template <typename T>
+void ColumnDecimal<T>::sort_column(const ColumnSorter* sorter, EqualFlags& flags,
+                                   IColumn::Permutation& perms, EqualRange& range,
+                                   bool last_column) const {
+    sorter->template sort_column(static_cast<const Self&>(*this), flags, perms, range, last_column);
+}
+
 template <>
 Decimal32 ColumnDecimal<Decimal32>::get_scale_multiplier() const {
     return common::exp10_i32(scale);
diff --git a/be/src/vec/columns/column_decimal.h b/be/src/vec/columns/column_decimal.h
index 2b3dac0ae5..81e037278c 100644
--- a/be/src/vec/columns/column_decimal.h
+++ b/be/src/vec/columns/column_decimal.h
@@ -226,6 +226,9 @@ public:
         data[self_row] = T();
     }
 
+    void sort_column(const ColumnSorter* sorter, EqualFlags& flags, IColumn::Permutation& perms,
+                     EqualRange& range, bool last_column) const override;
+
     UInt32 get_scale() const { return scale; }
 
     T get_scale_multiplier() const;
diff --git a/be/src/vec/columns/column_nullable.cpp b/be/src/vec/columns/column_nullable.cpp
index cff4d0f7c3..4f18163479 100644
--- a/be/src/vec/columns/column_nullable.cpp
+++ b/be/src/vec/columns/column_nullable.cpp
@@ -27,6 +27,7 @@
 #include "vec/common/nan_utils.h"
 #include "vec/common/sip_hash.h"
 #include "vec/common/typeid_cast.h"
+#include "vec/core/sort_block.h"
 
 namespace doris::vectorized {
 
@@ -508,6 +509,13 @@ void ColumnNullable::check_consistency() const {
     }
 }
 
+void ColumnNullable::sort_column(const ColumnSorter* sorter, EqualFlags& flags,
+                                 IColumn::Permutation& perms, EqualRange& range,
+                                 bool last_column) const {
+    sorter->sort_column(static_cast<const ColumnNullable&>(*this), flags, perms, range,
+                        last_column);
+}
+
 ColumnPtr make_nullable(const ColumnPtr& column, bool is_nullable) {
     if (is_column_nullable(*column)) return column;
 
diff --git a/be/src/vec/columns/column_nullable.h b/be/src/vec/columns/column_nullable.h
index 397f3be1bd..2a8bf52dd2 100644
--- a/be/src/vec/columns/column_nullable.h
+++ b/be/src/vec/columns/column_nullable.h
@@ -305,6 +305,9 @@ public:
         get_nested_column().generate_hash_values_for_runtime_filter();
     }
 
+    void sort_column(const ColumnSorter* sorter, EqualFlags& flags, IColumn::Permutation& perms,
+                     EqualRange& range, bool last_column) const override;
+
 private:
     WrappedPtr nested_column;
     WrappedPtr null_map;
diff --git a/be/src/vec/columns/column_string.cpp b/be/src/vec/columns/column_string.cpp
index c8b99e8ffa..18f0d74d23 100644
--- a/be/src/vec/columns/column_string.cpp
+++ b/be/src/vec/columns/column_string.cpp
@@ -25,6 +25,7 @@
 #include "vec/common/assert_cast.h"
 #include "vec/common/memcmp_small.h"
 #include "vec/common/unaligned.h"
+#include "vec/core/sort_block.h"
 
 namespace doris::vectorized {
 
@@ -424,6 +425,12 @@ void ColumnString::get_extremes(Field& min, Field& max) const {
     get(max_idx, max);
 }
 
+void ColumnString::sort_column(const ColumnSorter* sorter, EqualFlags& flags,
+                               IColumn::Permutation& perms, EqualRange& range,
+                               bool last_column) const {
+    sorter->sort_column(static_cast<const ColumnString&>(*this), flags, perms, range, last_column);
+}
+
 void ColumnString::protect() {
     get_chars().protect();
     get_offsets().protect();
diff --git a/be/src/vec/columns/column_string.h b/be/src/vec/columns/column_string.h
index d2744e20b7..8c504c6d08 100644
--- a/be/src/vec/columns/column_string.h
+++ b/be/src/vec/columns/column_string.h
@@ -265,6 +265,9 @@ public:
 
     ColumnPtr permute(const Permutation& perm, size_t limit) const override;
 
+    void sort_column(const ColumnSorter* sorter, EqualFlags& flags, IColumn::Permutation& perms,
+                     EqualRange& range, bool last_column) const override;
+
     //    ColumnPtr index(const IColumn & indexes, size_t limit) const override;
 
     template <typename Type>
diff --git a/be/src/vec/columns/column_vector.cpp b/be/src/vec/columns/column_vector.cpp
index 13c5810026..d60f8816a1 100644
--- a/be/src/vec/columns/column_vector.cpp
+++ b/be/src/vec/columns/column_vector.cpp
@@ -35,6 +35,7 @@
 #include "vec/common/nan_utils.h"
 #include "vec/common/sip_hash.h"
 #include "vec/common/unaligned.h"
+#include "vec/core/sort_block.h"
 
 namespace doris::vectorized {
 
@@ -111,6 +112,13 @@ void ColumnVector<T>::update_hashes_with_value(std::vector<SipHash>& hashes,
     SIP_HASHES_FUNCTION_COLUMN_IMPL();
 }
 
+template <typename T>
+void ColumnVector<T>::sort_column(const ColumnSorter* sorter, EqualFlags& flags,
+                                  IColumn::Permutation& perms, EqualRange& range,
+                                  bool last_column) const {
+    sorter->template sort_column(static_cast<const Self&>(*this), flags, perms, range, last_column);
+}
+
 template <typename T>
 struct ColumnVector<T>::less {
     const Self& parent;
diff --git a/be/src/vec/columns/column_vector.h b/be/src/vec/columns/column_vector.h
index 5101894b92..3a8ec82382 100644
--- a/be/src/vec/columns/column_vector.h
+++ b/be/src/vec/columns/column_vector.h
@@ -384,6 +384,9 @@ public:
         data[self_row] = T();
     }
 
+    void sort_column(const ColumnSorter* sorter, EqualFlags& flags, IColumn::Permutation& perms,
+                     EqualRange& range, bool last_column) const override;
+
 protected:
     Container data;
 };
diff --git a/be/src/vec/core/sort_block.cpp b/be/src/vec/core/sort_block.cpp
index d26566e1f9..657fd58d23 100644
--- a/be/src/vec/core/sort_block.cpp
+++ b/be/src/vec/core/sort_block.cpp
@@ -20,8 +20,6 @@
 
 #include "vec/core/sort_block.h"
 
-#include <pdqsort.h>
-
 #include "vec/columns/column_string.h"
 #include "vec/common/typeid_cast.h"
 
@@ -100,12 +98,12 @@ void sort_block(Block& block, const SortDescription& description, UInt64 limit)
         ColumnsWithSortDescriptions columns_with_sort_desc =
                 get_columns_with_sort_description(block, description);
         {
-            PartialSortingLess less(columns_with_sort_desc);
+            EqualFlags flags(size, 1);
+            EqualRange range {0, size};
 
-            if (limit) {
-                std::partial_sort(perm.begin(), perm.begin() + limit, perm.end(), less);
-            } else {
-                pdqsort(perm.begin(), perm.end(), less);
+            for (size_t i = 0; i < columns_with_sort_desc.size(); i++) {
+                ColumnSorter sorter(columns_with_sort_desc[i], limit);
+                sorter.operator()(flags, perm, range, i == columns_with_sort_desc.size() - 1);
             }
         }
 
diff --git a/be/src/vec/core/sort_block.h b/be/src/vec/core/sort_block.h
index bbdbcdb783..2a2babf46f 100644
--- a/be/src/vec/core/sort_block.h
+++ b/be/src/vec/core/sort_block.h
@@ -19,7 +19,9 @@
 // and modified by Doris
 
 #pragma once
+#include <pdqsort.h>
 
+#include "util/simd/bits.h"
 #include "vec/core/block.h"
 #include "vec/core/sort_description.h"
 
@@ -46,9 +48,412 @@ void stable_get_permutation(const Block& block, const SortDescription& descripti
   */
 bool is_already_sorted(const Block& block, const SortDescription& description);
 
-using ColumnsWithSortDescriptions = std::vector<std::pair<const IColumn*, SortColumnDescription>>;
+using ColumnWithSortDescription = std::pair<const IColumn*, SortColumnDescription>;
+
+using ColumnsWithSortDescriptions = std::vector<ColumnWithSortDescription>;
 
 ColumnsWithSortDescriptions get_columns_with_sort_description(const Block& block,
                                                               const SortDescription& description);
 
+struct EqualRangeIterator {
+    int range_begin;
+    int range_end;
+
+    EqualRangeIterator(const EqualFlags& flags) : EqualRangeIterator(flags, 0, flags.size()) {}
+
+    EqualRangeIterator(const EqualFlags& flags, int begin, int end) : _flags(flags), _end(end) {
+        range_begin = begin;
+        range_end = end;
+        _cur_range_begin = begin;
+        _cur_range_end = end;
+    }
+
+    bool next() {
+        if (_cur_range_begin >= _end) {
+            return false;
+        }
+
+        // `_flags[i]=1` indicates that the i-th row is equal to the previous row, which means we
+        // should continue to sort this row according to current column. Using the first non-zero
+        // value and first zero value after first non-zero value as two bounds, we can get an equal range here
+        if (!(_cur_range_begin == 0) || !(_flags[_cur_range_begin] == 1)) {
+            _cur_range_begin = simd::find_nonzero(_flags, _cur_range_begin + 1);
+            if (_cur_range_begin >= _end) {
+                return false;
+            }
+            _cur_range_begin--;
+        }
+
+        _cur_range_end = simd::find_zero(_flags, _cur_range_begin + 1);
+        DCHECK(_cur_range_end <= _end);
+
+        if (_cur_range_begin >= _cur_range_end) {
+            return false;
+        }
+
+        range_begin = _cur_range_begin;
+        range_end = _cur_range_end;
+        _cur_range_begin = _cur_range_end;
+        return true;
+    }
+
+private:
+    int _cur_range_begin;
+    int _cur_range_end;
+
+    const EqualFlags& _flags;
+    const int _end;
+};
+
+struct ColumnPartialSortingLess {
+    const ColumnWithSortDescription& _column_with_sort_desc;
+
+    explicit ColumnPartialSortingLess(const ColumnWithSortDescription& column)
+            : _column_with_sort_desc(column) {}
+
+    bool operator()(size_t a, size_t b) const {
+        int res = _column_with_sort_desc.second.direction *
+                  _column_with_sort_desc.first->compare_at(
+                          a, b, *_column_with_sort_desc.first,
+                          _column_with_sort_desc.second.nulls_direction);
+        if (res < 0) {
+            return true;
+        } else if (res > 0) {
+            return false;
+        }
+        return false;
+    }
+};
+
+template <typename T>
+struct PermutationWithInlineValue {
+    T inline_value;
+    uint32_t row_id;
+};
+
+template <typename T>
+using PermutationForColumn = std::vector<PermutationWithInlineValue<T>>;
+
+class ColumnSorter {
+public:
+    explicit ColumnSorter(const ColumnWithSortDescription& column, const int limit)
+            : _column_with_sort_desc(column),
+              _limit(limit),
+              _nulls_direction(column.second.nulls_direction),
+              _direction(column.second.direction) {}
+
+    void operator()(EqualFlags& flags, IColumn::Permutation& perms, EqualRange& range,
+                    bool last_column) const {
+        _column_with_sort_desc.first->sort_column(this, flags, perms, range, last_column);
+    }
+
+    void sort_column(const IColumn& column, EqualFlags& flags, IColumn::Permutation& perms,
+                     EqualRange& range, bool last_column) const {
+        int new_limit = _limit;
+        auto comparator = [&](const size_t a, const size_t b) {
+            return column.compare_at(a, b, *_column_with_sort_desc.first, _nulls_direction);
+        };
+        ColumnPartialSortingLess less(_column_with_sort_desc);
+        auto do_sort = [&](size_t first_iter, size_t last_iter) {
+            auto begin = perms.begin() + first_iter;
+            auto end = perms.begin() + last_iter;
+
+            if (UNLIKELY(_limit > 0 && first_iter < _limit && _limit <= last_iter)) {
+                int n = _limit - first_iter;
+                std::partial_sort(begin, begin + n, end, less);
+
+                auto nth = perms[_limit - 1];
+                size_t equal_count = 0;
+                for (auto iter = begin + n; iter < end; iter++) {
+                    if (comparator(*iter, nth) == 0) {
+                        std::iter_swap(iter, begin + n + equal_count);
+                        equal_count++;
+                    }
+                }
+                new_limit = _limit + equal_count;
+            } else {
+                pdqsort(begin, end, less);
+            }
+        };
+
+        EqualRangeIterator iterator(flags, range.first, range.second);
+        while (iterator.next()) {
+            int range_begin = iterator.range_begin;
+            int range_end = iterator.range_end;
+
+            if (UNLIKELY(_limit > 0 && range_begin > _limit)) {
+                break;
+            }
+            if (LIKELY(range_end - range_begin > 1)) {
+                do_sort(range_begin, range_end);
+                if (!last_column) {
+                    flags[range_begin] = 0;
+                    for (int i = range_begin + 1; i < range_end; i++) {
+                        flags[i] &= comparator(perms[i - 1], perms[i]) == 0;
+                    }
+                }
+            }
+        }
+        _shrink_to_fit(perms, flags, new_limit);
+    }
+
+    template <template <typename type> typename ColumnType, typename T>
+    void sort_column(const ColumnType<T>& column, EqualFlags& flags, IColumn::Permutation& perms,
+                     EqualRange& range, bool last_column) const {
+        if (!_should_inline_value(perms)) {
+            _sort_by_default(column, flags, perms, range, last_column);
+        } else {
+            _sort_by_inlined_permutation<T>(column, flags, perms, range, last_column);
+        }
+    }
+
+    void sort_column(const ColumnString& column, EqualFlags& flags, IColumn::Permutation& perms,
+                     EqualRange& range, bool last_column) const {
+        if (!_should_inline_value(perms)) {
+            _sort_by_default(column, flags, perms, range, last_column);
+        } else {
+            _sort_by_inlined_permutation<StringRef>(column, flags, perms, range, last_column);
+        }
+    }
+
+    void sort_column(const ColumnNullable& column, EqualFlags& flags, IColumn::Permutation& perms,
+                     EqualRange& range, bool last_column) const {
+        if (!column.has_null()) {
+            column.get_nested_column().sort_column(this, flags, perms, range, last_column);
+        } else {
+            const auto& null_map = column.get_null_map_data();
+            EqualRangeIterator iterator(flags, range.first, range.second);
+            while (iterator.next()) {
+                int range_begin = iterator.range_begin;
+                int range_end = iterator.range_end;
+
+                if (UNLIKELY(_limit > 0 && range_begin > _limit)) {
+                    break;
+                }
+                bool null_first = _nulls_direction * _direction < 0;
+                if (LIKELY(range_end - range_begin > 1)) {
+                    int range_split = 0;
+                    if (null_first) {
+                        range_split = std::partition(perms.begin() + range_begin,
+                                                     perms.begin() + range_end,
+                                                     [&](size_t row_id) -> bool {
+                                                         return null_map[row_id] != 0;
+                                                     }) -
+                                      perms.begin();
+                    } else {
+                        range_split = std::partition(perms.begin() + range_begin,
+                                                     perms.begin() + range_end,
+                                                     [&](size_t row_id) -> bool {
+                                                         return null_map[row_id] == 0;
+                                                     }) -
+                                      perms.begin();
+                    }
+                    std::pair<size_t, size_t> is_null_range = {range_begin, range_split};
+                    std::pair<size_t, size_t> not_null_range = {range_split, range_end};
+                    if (!null_first) {
+                        std::swap(is_null_range, not_null_range);
+                    }
+
+                    if (not_null_range.first < not_null_range.second) {
+                        flags[not_null_range.first] = 0;
+                    }
+                    if (is_null_range.first < is_null_range.second) {
+                        std::fill(flags.begin() + is_null_range.first,
+                                  flags.begin() + is_null_range.second, 1);
+
+                        flags[is_null_range.first] = 0;
+                    }
+                }
+            }
+
+            column.get_nested_column().sort_column(this, flags, perms, range, last_column);
+        }
+    }
+
+private:
+    bool _should_inline_value(const IColumn::Permutation& perms) const {
+        return _limit == 0 || _limit > (perms.size() / 5);
+    }
+
+    template <typename T>
+    void _shrink_to_fit(PermutationForColumn<T>& permutation_for_column,
+                        IColumn::Permutation& perms, EqualFlags& flags, int limit) const {
+        if (limit < perms.size() && limit != 0) {
+            permutation_for_column.resize(limit);
+            perms.resize(limit);
+            flags.resize(limit);
+        }
+    }
+
+    void _shrink_to_fit(IColumn::Permutation& perms, EqualFlags& flags, int limit) const {
+        if (_limit < perms.size() && limit != 0) {
+            perms.resize(limit);
+            flags.resize(limit);
+        }
+    }
+
+    template <typename ColumnType>
+    static constexpr bool always_false_v = false;
+
+    template <typename ColumnType, typename T>
+    void _create_permutation(const ColumnType& column,
+                             PermutationWithInlineValue<T>* __restrict permutation_for_column,
+                             const IColumn::Permutation& perms) const {
+        for (size_t i = 0; i < perms.size(); i++) {
+            size_t row_id = perms[i];
+            if constexpr (std::is_same_v<ColumnType, ColumnVector<T>> ||
+                          std::is_same_v<ColumnType, ColumnDecimal<T>>) {
+                permutation_for_column[i].inline_value = column.get_data()[row_id];
+            } else if constexpr (std::is_same_v<ColumnType, ColumnString>) {
+                permutation_for_column[i].inline_value = column.get_data_at(row_id);
+            } else {
+                static_assert(always_false_v<ColumnType>);
+            }
+            permutation_for_column[i].row_id = row_id;
+        }
+    }
+
+    template <typename ColumnType>
+    void _sort_by_default(const ColumnType& column, EqualFlags& flags, IColumn::Permutation& perms,
+                          EqualRange& range, bool last_column) const {
+        int new_limit = _limit;
+        auto comparator = [&](const size_t a, const size_t b) {
+            if constexpr (!std::is_same_v<ColumnType, ColumnString>) {
+                auto value_a = column.get_data()[a];
+                auto value_b = column.get_data()[b];
+                return value_a > value_b ? 1 : (value_a < value_b ? -1 : 0);
+            } else {
+                return column.compare_at(a, b, column, _nulls_direction);
+            }
+        };
+
+        auto sort_comparator = [&](const size_t a, const size_t b) {
+            return comparator(a, b) * _direction < 0;
+        };
+        auto do_sort = [&](size_t first_iter, size_t last_iter) {
+            auto begin = perms.begin() + first_iter;
+            auto end = perms.begin() + last_iter;
+
+            if (UNLIKELY(_limit > 0 && first_iter < _limit && _limit <= last_iter)) {
+                int n = _limit - first_iter;
+                std::partial_sort(begin, begin + n, end, sort_comparator);
+
+                auto nth = perms[_limit - 1];
+                size_t equal_count = 0;
+                for (auto iter = begin + n; iter < end; iter++) {
+                    if (comparator(*iter, nth) == 0) {
+                        std::iter_swap(iter, begin + n + equal_count);
+                        equal_count++;
+                    }
+                }
+                new_limit = _limit + equal_count;
+            } else {
+                pdqsort(begin, end, sort_comparator);
+            }
+        };
+
+        EqualRangeIterator iterator(flags, range.first, range.second);
+        while (iterator.next()) {
+            int range_begin = iterator.range_begin;
+            int range_end = iterator.range_end;
+
+            if (UNLIKELY(_limit > 0 && range_begin > _limit)) {
+                break;
+            }
+            if (LIKELY(range_end - range_begin > 1)) {
+                do_sort(range_begin, range_end);
+                if (!last_column) {
+                    flags[range_begin] = 0;
+                    for (int i = range_begin + 1; i < range_end; i++) {
+                        flags[i] &= comparator(perms[i - 1], perms[i]) == 0;
+                    }
+                }
+            }
+        }
+        _shrink_to_fit(perms, flags, new_limit);
+    }
+
+    template <typename InlineType, typename ColumnType>
+    void _sort_by_inlined_permutation(const ColumnType& column, EqualFlags& flags,
+                                      IColumn::Permutation& perms, EqualRange& range,
+                                      bool last_column) const {
+        int new_limit = _limit;
+        // create inlined permutation
+        PermutationForColumn<InlineType> permutation_for_column(perms.size());
+        _create_permutation(column, permutation_for_column.data(), perms);
+        auto comparator = [&](const PermutationWithInlineValue<InlineType>& a,
+                              const PermutationWithInlineValue<InlineType>& b) {
+            if constexpr (!std::is_same_v<ColumnType, ColumnString>) {
+                return a.inline_value > b.inline_value ? 1
+                                                       : (a.inline_value < b.inline_value ? -1 : 0);
+            } else {
+                return memcmp_small_allow_overflow15(a.inline_value.data, a.inline_value.size,
+                                                     b.inline_value.data, b.inline_value.size);
+            }
+        };
+
+        auto sort_comparator = [&](const PermutationWithInlineValue<InlineType>& a,
+                                   const PermutationWithInlineValue<InlineType>& b) {
+            return comparator(a, b) * _direction < 0;
+        };
+        auto do_sort = [&](size_t first_iter, size_t last_iter) {
+            auto begin = permutation_for_column.begin() + first_iter;
+            auto end = permutation_for_column.begin() + last_iter;
+
+            if (UNLIKELY(_limit > 0 && first_iter < _limit && _limit <= last_iter)) {
+                int n = _limit - first_iter;
+                std::partial_sort(begin, begin + n, end, sort_comparator);
+
+                auto nth = permutation_for_column[_limit - 1];
+                size_t equal_count = 0;
+                for (auto iter = begin + n; iter < end; iter++) {
+                    if (comparator(*iter, nth) == 0) {
+                        std::iter_swap(iter, begin + n + equal_count);
+                        equal_count++;
+                    }
+                }
+                new_limit = _limit + equal_count;
+            } else {
+                pdqsort(begin, end, sort_comparator);
+            }
+        };
+
+        EqualRangeIterator iterator(flags, range.first, range.second);
+        while (iterator.next()) {
+            int range_begin = iterator.range_begin;
+            int range_end = iterator.range_end;
+
+            if (UNLIKELY(_limit > 0 && range_begin > _limit)) {
+                break;
+            }
+            if (LIKELY(range_end - range_begin > 1)) {
+                do_sort(range_begin, range_end);
+                if (!last_column) {
+                    flags[range_begin] = 0;
+                    for (int i = range_begin + 1; i < range_end; i++) {
+                        flags[i] &= comparator(permutation_for_column[i - 1],
+                                               permutation_for_column[i]) == 0;
+                    }
+                }
+            }
+        }
+        _shrink_to_fit(permutation_for_column, perms, flags, new_limit);
+        // restore `perms` from `permutation_for_column`
+        _restore_permutation(permutation_for_column, perms.data());
+    }
+
+    template <typename T>
+    void _restore_permutation(const PermutationForColumn<T>& permutation_for_column,
+                              size_t* __restrict perms) const {
+        for (size_t i = 0; i < permutation_for_column.size(); i++) {
+            perms[i] = permutation_for_column[i].row_id;
+        }
+    }
+
+    const ColumnWithSortDescription& _column_with_sort_desc;
+    const int _limit;
+    const int _nulls_direction;
+    const int _direction;
+};
+
 } // namespace doris::vectorized
diff --git a/be/src/vec/exec/vsort_node.cpp b/be/src/vec/exec/vsort_node.cpp
index f12a436d1d..7231016523 100644
--- a/be/src/vec/exec/vsort_node.cpp
+++ b/be/src/vec/exec/vsort_node.cpp
@@ -48,6 +48,7 @@ Status VSortNode::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(_vsort_exec_exprs.prepare(state, child(0)->row_desc(), _row_descriptor));
     _unsorted_block.reset(new MutableBlock(
             VectorizedUtils::create_empty_columnswithtypename(child(0)->row_desc())));
+    _partial_sort_timer = ADD_TIMER(_runtime_profile, "PartialSortTime");
     return Status::OK();
 }
 
@@ -141,7 +142,10 @@ Status VSortNode::sort_input(RuntimeState* state) {
         if (_unsorted_block->rows() > 0) {
             _total_mem_usage += _unsorted_block->allocated_bytes();
             Block block = _unsorted_block->to_block(0);
-            RETURN_IF_ERROR(partial_sort(block));
+            {
+                SCOPED_TIMER(_partial_sort_timer);
+                RETURN_IF_ERROR(partial_sort(block));
+            }
             // dispose TOP-N logic
             if (_limit != -1) {
                 // Here is a little opt to reduce the mem uasge, we build a max heap
diff --git a/be/src/vec/exec/vsort_node.h b/be/src/vec/exec/vsort_node.h
index 1565be5ca6..717059787b 100644
--- a/be/src/vec/exec/vsort_node.h
+++ b/be/src/vec/exec/vsort_node.h
@@ -86,6 +86,7 @@ private:
     std::priority_queue<SortBlockCursor> _block_priority_queue;
 
     std::unique_ptr<MutableBlock> _unsorted_block;
+    RuntimeProfile::Counter* _partial_sort_timer = nullptr;
 
     static constexpr size_t BUFFERED_BLOCK_SIZE = 1024 * 1024;
     static constexpr size_t BUFFERED_BLOCK_BYTES = 16 << 20;