[Feature](Vectorization) support function topn && remove some unused code

be/src/exprs/topn_function.cpp

@@ -91,50 +91,10 @@ StringVal TopNFunctions::topn_finalize(FunctionContext* ctx, const StringVal& sr
     return result;
 }
 
-template void TopNFunctions::topn_update(FunctionContext*, const BooleanVal&, const IntVal&,
-                                         StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const TinyIntVal&, const IntVal&,
-                                         StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const SmallIntVal&, const IntVal&,
-                                         StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const IntVal&, const IntVal&,
-                                         StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const BigIntVal&, const IntVal&,
-                                         StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const FloatVal&, const IntVal&,
-                                         StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const DoubleVal&, const IntVal&,
-                                         StringVal*);
 template void TopNFunctions::topn_update(FunctionContext*, const StringVal&, const IntVal&,
                                          StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const DateTimeVal&, const IntVal&,
-                                         StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const LargeIntVal&, const IntVal&,
-                                         StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const DecimalV2Val&, const IntVal&,
-                                         StringVal*);
 
-template void TopNFunctions::topn_update(FunctionContext*, const BooleanVal&, const IntVal&,
-                                         const IntVal&, StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const TinyIntVal&, const IntVal&,
-                                         const IntVal&, StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const SmallIntVal&, const IntVal&,
-                                         const IntVal&, StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const IntVal&, const IntVal&,
-                                         const IntVal&, StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const BigIntVal&, const IntVal&,
-                                         const IntVal&, StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const FloatVal&, const IntVal&,
-                                         const IntVal&, StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const DoubleVal&, const IntVal&,
-                                         const IntVal&, StringVal*);
 template void TopNFunctions::topn_update(FunctionContext*, const StringVal&, const IntVal&,
                                          const IntVal&, StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const DateTimeVal&, const IntVal&,
-                                         const IntVal&, StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const LargeIntVal&, const IntVal&,
-                                         const IntVal&, StringVal*);
-template void TopNFunctions::topn_update(FunctionContext*, const DecimalV2Val&, const IntVal&,
-                                         const IntVal&, StringVal*);
 
 } // namespace doris

be/src/vec/CMakeLists.txt

@@ -32,6 +32,7 @@ set(VEC_FILES
   aggregate_functions/aggregate_function_reader.cpp
   aggregate_functions/aggregate_function_window.cpp
   aggregate_functions/aggregate_function_stddev.cpp
+  aggregate_functions/aggregate_function_topn.cpp
   aggregate_functions/aggregate_function_simple_factory.cpp
   columns/collator.cpp
   columns/column.cpp

be/src/vec/aggregate_functions/aggregate_function.h

@@ -20,13 +20,6 @@
 
 #pragma once
 
-#include <cstddef>
-#include <istream>
-#include <memory>
-#include <ostream>
-#include <type_traits>
-#include <vector>
-
 #include "vec/common/exception.h"
 #include "vec/core/block.h"
 #include "vec/core/column_numbers.h"
@@ -95,13 +88,15 @@ class IAggregateFunction {
                      Arena* arena) const = 0;
 
     /// Merges state (on which place points to) with other state of current aggregation function.
-    virtual void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena* arena) const = 0;
+    virtual void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
+                       Arena* arena) const = 0;
 
     /// Serializes state (to transmit it over the network, for example).
     virtual void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const = 0;
 
     /// Deserializes state. This function is called only for empty (just created) states.
-    virtual void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf, Arena* arena) const = 0;
+    virtual void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
+                             Arena* arena) const = 0;
 
     /// Returns true if a function requires Arena to handle own states (see add(), merge(), deserialize()).
     virtual bool allocates_memory_in_arena() const { return false; }
@@ -114,21 +109,11 @@ class IAggregateFunction {
       */
     virtual bool is_state() const { return false; }
 
-    /// if return false, during insert_result_into function, you colud get nullable result column, 
+    /// if return false, during insert_result_into function, you colud get nullable result column,
     /// so could insert to null value by yourself, rather than by AggregateFunctionNullBase;
     /// because you maybe be calculate a invalid value, but want to use null replace it;
     virtual bool insert_to_null_default() const { return true; }
 
-    /** The inner loop that uses the function pointer is better than using the virtual function.
-      * The reason is that in the case of virtual functions GCC 5.1.2 generates code,
-      *  which, at each iteration of the loop, reloads the function address (the offset value in the virtual function table) from memory to the register.
-      * This gives a performance drop on simple queries around 12%.
-      * After the appearance of better compilers, the code can be removed.
-      */
-    using AddFunc = void (*)(const IAggregateFunction*, AggregateDataPtr, const IColumn**, size_t,
-                             Arena*);
-    virtual AddFunc get_address_of_add_function() const = 0;
-
     /** Contains a loop with calls to "add" function. You can collect arguments into array "places"
       *  and do a single call to "add_batch" for devirtualization and inlining.
       */
@@ -150,12 +135,6 @@ class IAggregateFunction {
                                         AggregateDataPtr place, const IColumn** columns,
                                         Arena* arena) const = 0;
 
-    /** This is used for runtime code generation to determine, which header files to include in generated source.
-      * Always implement it as
-      * const char * get_header_file_path() const override { return __FILE__; }
-      */
-    virtual const char* get_header_file_path() const = 0;
-
     const DataTypes& get_argument_types() const { return argument_types; }
     const Array& get_parameters() const { return parameters; }
 
@@ -167,18 +146,10 @@ class IAggregateFunction {
 /// Implement method to obtain an address of 'add' function.
 template <typename Derived>
 class IAggregateFunctionHelper : public IAggregateFunction {
-private:
-    static void add_free(const IAggregateFunction* that, AggregateDataPtr place,
-                         const IColumn** columns, size_t row_num, Arena* arena) {
-        static_cast<const Derived&>(*that).add(place, columns, row_num, arena);
-    }
-
 public:
     IAggregateFunctionHelper(const DataTypes& argument_types_, const Array& parameters_)
             : IAggregateFunction(argument_types_, parameters_) {}
 
-    AddFunc get_address_of_add_function() const override { return &add_free; }
-
     void add_batch(size_t batch_size, AggregateDataPtr* places, size_t place_offset,
                    const IColumn** columns, Arena* arena) const override {
         for (size_t i = 0; i < batch_size; ++i)

be/src/vec/aggregate_functions/aggregate_function_avg.h

@@ -101,7 +101,8 @@ class AggregateFunctionAvg final
         this->data(place).count = 0;
     }
 
-    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena*) const override {
+    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
+               Arena*) const override {
         this->data(place).sum += this->data(rhs).sum;
         this->data(place).count += this->data(rhs).count;
     }
@@ -110,7 +111,8 @@ class AggregateFunctionAvg final
         this->data(place).write(buf);
     }
 
-    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf, Arena*) const override {
+    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
+                     Arena*) const override {
         this->data(place).read(buf);
     }
 
@@ -119,8 +121,6 @@ class AggregateFunctionAvg final
         column.get_data().push_back(this->data(place).template result<ResultType>());
     }
 
-    const char* get_header_file_path() const override { return __FILE__; }
-
 private:
     UInt32 scale;
 };

be/src/vec/aggregate_functions/aggregate_function_bitmap.h

@@ -16,8 +16,6 @@
 // under the License.
 
 #pragma once
-#include <istream>
-#include <ostream>
 
 #include "vec/aggregate_functions/aggregate_function.h"
 #include "vec/columns/column_complex.h"
@@ -91,7 +89,8 @@ class AggregateFunctionBitmapOp final
         this->data(place).add(column.get_data()[row_num]);
     }
 
-    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena*) const override {
+    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
+               Arena*) const override {
         this->data(place).merge(
                 const_cast<AggregateFunctionBitmapData<Op>&>(this->data(rhs)).get());
     }
@@ -100,7 +99,8 @@ class AggregateFunctionBitmapOp final
         this->data(place).write(buf);
     }
 
-    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf, Arena*) const override {
+    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
+                     Arena*) const override {
         this->data(place).read(buf);
     }
 
@@ -109,8 +109,6 @@ class AggregateFunctionBitmapOp final
         column.get_data().push_back(
                 const_cast<AggregateFunctionBitmapData<Op>&>(this->data(place)).get());
     }
-
-    const char* get_header_file_path() const override { return __FILE__; }
 };
 
 template <bool nullable, typename ColVecType>
@@ -146,15 +144,17 @@ class AggregateFunctionBitmapCount final
         }
     }
 
-    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena*) const override {
+    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
+               Arena*) const override {
         this->data(place).merge(const_cast<AggFunctionData&>(this->data(rhs)).get());
     }
 
     void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
         this->data(place).write(buf);
     }
 
-    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf, Arena*) const override {
+    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
+                     Arena*) const override {
         this->data(place).read(buf);
     }
 
@@ -163,8 +163,6 @@ class AggregateFunctionBitmapCount final
         auto& column = static_cast<ColVecResult&>(to);
         column.get_data().push_back(value_data.cardinality());
     }
-
-    const char* get_header_file_path() const override { return __FILE__; }
 };
 
 AggregateFunctionPtr create_aggregate_function_bitmap_union(const std::string& name,

be/src/vec/aggregate_functions/aggregate_function_count.h

@@ -50,27 +50,25 @@ class AggregateFunctionCount final
         ++data(place).count;
     }
 
-    void reset(AggregateDataPtr place) const override {
-        this->data(place).count = 0;
-    }
+    void reset(AggregateDataPtr place) const override { this->data(place).count = 0; }
 
-    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena*) const override {
+    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
+               Arena*) const override {
         data(place).count += data(rhs).count;
     }
 
     void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
         write_var_uint(data(place).count, buf);
     }
 
-    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf, Arena*) const override {
+    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
+                     Arena*) const override {
         read_var_uint(data(place).count, buf);
     }
 
     void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
         assert_cast<ColumnInt64&>(to).get_data().push_back(data(place).count);
     }
-
-    const char* get_header_file_path() const override { return __FILE__; }
 };
 
 /// Simply count number of not-NULL values.
@@ -90,33 +88,33 @@ class AggregateFunctionCountNotNullUnary final
         data(place).count += !assert_cast<const ColumnNullable&>(*columns[0]).is_null_at(row_num);
     }
 
-    void reset(AggregateDataPtr place) const override {
-        data(place).count = 0;
-    }
+    void reset(AggregateDataPtr place) const override { data(place).count = 0; }
 
-    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena*) const override {
+    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
+               Arena*) const override {
         data(place).count += data(rhs).count;
     }
 
     void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
         write_var_uint(data(place).count, buf);
     }
 
-    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf, Arena*) const override {
+    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
+                     Arena*) const override {
         read_var_uint(data(place).count, buf);
     }
 
     void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
         if (to.is_nullable()) {
-            auto& null_column = assert_cast<ColumnNullable &>(to);
+            auto& null_column = assert_cast<ColumnNullable&>(to);
             null_column.get_null_map_data().push_back(0);
-            assert_cast<ColumnInt64 &>(null_column.get_nested_column()).get_data().push_back(data(place).count);
+            assert_cast<ColumnInt64&>(null_column.get_nested_column())
+                    .get_data()
+                    .push_back(data(place).count);
         } else {
-            assert_cast<ColumnInt64 &>(to).get_data().push_back(data(place).count);
+            assert_cast<ColumnInt64&>(to).get_data().push_back(data(place).count);
         }
     }
-
-    const char* get_header_file_path() const override { return __FILE__; }
 };
 
 } // namespace doris::vectorized

be/src/vec/aggregate_functions/aggregate_function_distinct.h

@@ -69,13 +69,12 @@ struct AggregateFunctionDistinctGenericData {
         Set::LookupResult it;
         bool inserted;
         for (const auto& elem : rhs.set)
-            set.emplace(ArenaKeyHolder{elem.get_value(), *arena}, it, inserted);
+            set.emplace(ArenaKeyHolder {elem.get_value(), *arena}, it, inserted);
     }
 
     void serialize(BufferWritable& buf) const {
         write_var_uint(set.size(), buf);
-        for (const auto& elem : set)
-            write_string_binary(elem.get_value(), buf);
+        for (const auto& elem : set) write_string_binary(elem.get_value(), buf);
     }
 
     void deserialize(BufferReadable& buf, Arena* arena) {
@@ -121,7 +120,7 @@ struct AggregateFunctionDistinctMultipleGenericData : public AggregateFunctionDi
 
         Set::LookupResult it;
         bool inserted;
-        auto key_holder = SerializedKeyHolder{value, *arena};
+        auto key_holder = SerializedKeyHolder {value, *arena};
         set.emplace(key_holder, it, inserted);
     }
 
@@ -180,7 +179,8 @@ class AggregateFunctionDistinct
         this->data(place).serialize(buf);
     }
 
-    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf, Arena* arena) const override {
+    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
+                     Arena* arena) const override {
         this->data(place).deserialize(buf, arena);
     }
 
@@ -217,8 +217,6 @@ class AggregateFunctionDistinct
     DataTypePtr get_return_type() const override { return nested_func->get_return_type(); }
 
     bool allocates_memory_in_arena() const override { return true; }
-
-    const char* get_header_file_path() const override { return __FILE__; }
 };
 
 } // namespace doris::vectorized