[Enhancement] optimize performance of percentile_cont function (#14609)

(cherry picked from commit 4e3418c)

# Conflicts:
#	be/src/exprs/agg/percentile_cont.h

be/src/exprs/agg/percentile_cont.h

@@ -11,6 +11,7 @@
 #include "column/vectorized_fwd.h"
 #include "exprs/agg/aggregate.h"
 #include "gutil/casts.h"
+#include "util/orlp/pdqsort.h"
 
 namespace starrocks::vectorized {
 
@@ -56,16 +57,27 @@ class PercentileContAggregateFunction final
 
         const Slice slice = column->get(row_num).get_slice();
         double rate = *reinterpret_cast<double*>(slice.data);
-        size_t items_size = *reinterpret_cast<size_t*>(slice.data + sizeof(double));
+        size_t second_size = *reinterpret_cast<size_t*>(slice.data + sizeof(double));
         auto data_ptr = slice.data + sizeof(double) + sizeof(size_t);
 
+<<<<<<< HEAD
         vector<InputCppType> res;
         vector<InputCppType>& vec = this->data(state).items;
         res.resize(vec.size() + items_size);
+=======
+        auto second_start = reinterpret_cast<InputCppType*>(data_ptr);
+        auto second_end = reinterpret_cast<InputCppType*>(data_ptr + second_size * sizeof(InputCppType));
+
+        // TODO(murphy) reduce the copy overhead of merge algorithm
+        auto& output = this->data(state).items;
+        size_t first_size = output.size();
+        output.resize(first_size + second_size);
+        auto first_end = output.begin() + first_size;
+        std::copy(second_start, second_end, first_end);
+        // TODO: optimize it with SIMD bitonic merge
+        std::inplace_merge(output.begin(), first_end, output.end());
+>>>>>>> 4e3418cab ([Enhancement] optimize performance of percentile_cont function (#14609))
 
-        std::merge(vec.begin(), vec.end(), reinterpret_cast<InputCppType*>(data_ptr),
-                   reinterpret_cast<InputCppType*>(data_ptr + items_size * sizeof(InputCppType)), res.begin());
-        this->data(state).items = std::move(res);
         this->data(state).rate = rate;
     }
 
@@ -83,17 +95,17 @@ class PercentileContAggregateFunction final
         memcpy(bytes.data() + old_size + sizeof(double), &items_size, sizeof(size_t));
         memcpy(bytes.data() + old_size + sizeof(double) + sizeof(size_t), this->data(state).items.data(),
                items_size * sizeof(InputCppType));
-        std::sort(reinterpret_cast<InputCppType*>(bytes.data() + old_size + sizeof(double) + sizeof(size_t)),
-                  reinterpret_cast<InputCppType*>(bytes.data() + old_size + sizeof(double) + sizeof(size_t) +
-                                                  items_size * sizeof(InputCppType)));
+        pdqsort(false, reinterpret_cast<InputCppType*>(bytes.data() + old_size + sizeof(double) + sizeof(size_t)),
+                reinterpret_cast<InputCppType*>(bytes.data() + old_size + sizeof(double) + sizeof(size_t) +
+                                                items_size * sizeof(InputCppType)));
 
         column->get_offset().emplace_back(new_size);
     }
 
     void finalize_to_column(FunctionContext* ctx, ConstAggDataPtr __restrict state, Column* to) const override {
         using CppType = RunTimeCppType<PT>;
         std::vector<CppType> new_vector = this->data(state).items;
-        std::sort(new_vector.begin(), new_vector.end());
+        pdqsort(false, new_vector.begin(), new_vector.end());
         const double& rate = this->data(state).rate;
 
         ResultColumnType* column = down_cast<ResultColumnType*>(to);
@@ -134,6 +146,7 @@ class PercentileContAggregateFunction final
         Bytes& bytes = dst_column->get_bytes();
         double rate = ColumnHelper::get_const_value<TYPE_DOUBLE>(src[1]);
         InputColumnType src_column = *down_cast<const InputColumnType*>(src[0].get());
+<<<<<<< HEAD
         InputCppType* src_data = src_column.get_data().data();
         for (auto i = 0; i < chunk_size; ++i) {
             size_t old_size = bytes.size();
@@ -143,6 +156,16 @@ class PercentileContAggregateFunction final
             memcpy(bytes.data() + old_size + sizeof(double) + sizeof(size_t), &src_data[i], sizeof(InputCppType));
             dst_column->get_offset().push_back(bytes.size());
         }
+=======
+        pdqsort(false, src_column.get_data().begin(), src_column.get_data().end());
+
+        bytes.resize(old_size + sizeof(double) + sizeof(size_t) + chunk_size * sizeof(InputCppType));
+
+        memcpy(bytes.data() + old_size, &rate, sizeof(double));
+        memcpy(bytes.data() + old_size + sizeof(double), &chunk_size, sizeof(size_t));
+        memcpy(bytes.data() + old_size + sizeof(double) + sizeof(size_t), src_column.get_data().data(),
+               chunk_size * sizeof(InputCppType));
+>>>>>>> 4e3418cab ([Enhancement] optimize performance of percentile_cont function (#14609))
     }
 
     std::string get_name() const override { return "percentile_cont"; }