[improve](function) support any_value function with complex type (apache#49419)

Problem Summary:

support any_value function with complex type: array/map/struct
bitmap/hll/quantile/agg_state/json

Author: zhangstar333

be/src/vec/aggregate_functions/aggregate_function_min_max.cpp

@@ -20,14 +20,16 @@
 
 #include "vec/aggregate_functions/aggregate_function_min_max.h"
 
+#include "runtime/define_primitive_type.h"
 #include "vec/aggregate_functions/aggregate_function_simple_factory.h"
 #include "vec/aggregate_functions/factory_helpers.h"
 #include "vec/aggregate_functions/helpers.h"
 #include "vec/core/types.h"
+#include "vec/data_types/data_type.h"
 #include "vec/data_types/data_type_nullable.h"
 
 namespace doris::vectorized {
-/// min, max, any
+/// min, max
 template <template <typename> class Data>
 AggregateFunctionPtr create_aggregate_function_single_value(const String& name,
                                                             const DataTypes& argument_types,
@@ -88,13 +90,40 @@ AggregateFunctionPtr create_aggregate_function_single_value(const String& name,
     return nullptr;
 }
 
+// any_value
+template <template <typename> class Data>
+AggregateFunctionPtr create_aggregate_function_single_value_any_value_function(
+        const String& name, const DataTypes& argument_types, const bool result_is_nullable,
+        const AggregateFunctionAttr& attr) {
+    AggregateFunctionPtr res = create_aggregate_function_single_value<Data>(
+            name, argument_types, result_is_nullable, attr);
+    if (res) {
+        return res;
+    }
+    const DataTypePtr& argument_type = remove_nullable(argument_types[0]);
+    if (argument_type->get_primitive_type() == PrimitiveType::TYPE_ARRAY ||
+        argument_type->get_primitive_type() == PrimitiveType::TYPE_MAP ||
+        argument_type->get_primitive_type() == PrimitiveType::TYPE_STRUCT ||
+        argument_type->get_primitive_type() == PrimitiveType::TYPE_AGG_STATE ||
+        argument_type->get_primitive_type() == PrimitiveType::TYPE_OBJECT ||
+        argument_type->get_primitive_type() == PrimitiveType::TYPE_HLL ||
+        argument_type->get_primitive_type() == PrimitiveType::TYPE_QUANTILE_STATE) {
+        return creator_without_type::create<
+                AggregateFunctionsSingleValue<SingleValueDataComplexType>>(argument_types,
+                                                                           result_is_nullable);
+    }
+
+    return nullptr;
+}
+
 void register_aggregate_function_minmax(AggregateFunctionSimpleFactory& factory) {
     factory.register_function_both(
             "max", create_aggregate_function_single_value<AggregateFunctionMaxData>);
     factory.register_function_both(
             "min", create_aggregate_function_single_value<AggregateFunctionMinData>);
     factory.register_function_both(
-            "any", create_aggregate_function_single_value<AggregateFunctionAnyData>);
+            "any",
+            create_aggregate_function_single_value_any_value_function<AggregateFunctionAnyData>);
     factory.register_alias("any", "any_value");
 }
 

be/src/vec/aggregate_functions/aggregate_function_min_max.h

@@ -21,9 +21,12 @@
 #pragma once
 
 #include <fmt/format.h>
+#include <glog/logging.h>
 #include <string.h>
 
 #include <memory>
+#include <string>
+#include <type_traits>
 #include <vector>
 
 #include "common/logging.h"
@@ -519,18 +522,104 @@ struct AggregateFunctionMinData : Data {
     static const char* name() { return "min"; }
 };
 
+// this is used for plain type about any_value function
 template <typename Data>
 struct AggregateFunctionAnyData : Data {
     using Self = AggregateFunctionAnyData;
     using Data::IsFixedLength;
+    static const char* name() { return "any"; }
     constexpr static bool IS_ANY = true;
-
     void change_if_better(const IColumn& column, size_t row_num, Arena*) {
         this->change_first_time(column, row_num, nullptr);
     }
+
     void change_if_better(const Self& to, Arena*) { this->change_first_time(to, nullptr); }
+};
 
+// this is used for complex type about any_value function
+struct SingleValueDataComplexType {
     static const char* name() { return "any"; }
+    constexpr static bool IS_ANY = true;
+    constexpr static bool IsFixedLength = false;
+    using Self = SingleValueDataComplexType;
+
+    SingleValueDataComplexType() = default;
+
+    SingleValueDataComplexType(const DataTypes& argument_types, int be_version) {
+        column_type = argument_types[0];
+        column_data = column_type->create_column();
+        be_exec_version = be_version;
+    }
+
+    bool has() const { return has_value; }
+
+    void change_first_time(const IColumn& column, size_t row_num, Arena*) {
+        if (UNLIKELY(!has())) {
+            change_impl(column, row_num);
+        }
+    }
+
+    void change_first_time(const Self& to, Arena*) {
+        if (UNLIKELY(!has() && to.has())) {
+            change_impl(*to.column_data, 0);
+        }
+    }
+
+    void change_impl(const IColumn& column, size_t row_num) {
+        DCHECK_EQ(column_data->size(), 0);
+        column_data->insert_from(column, row_num);
+        has_value = true;
+    }
+
+    void insert_result_into(IColumn& to) const {
+        if (has()) {
+            to.insert_from(*column_data, 0);
+        } else {
+            to.insert_default();
+        }
+    }
+
+    void reset() {
+        column_data->clear();
+        has_value = false;
+    }
+
+    void write(BufferWritable& buf) const {
+        write_binary(has(), buf);
+        if (!has()) {
+            return;
+        }
+        auto size_bytes =
+                column_type->get_uncompressed_serialized_bytes(*column_data, be_exec_version);
+        std::string memory_buffer(size_bytes, '0');
+        auto* p = column_type->serialize(*column_data, memory_buffer.data(), be_exec_version);
+        write_binary(memory_buffer, buf);
+        DCHECK_EQ(p, memory_buffer.data() + size_bytes);
+    }
+
+    void read(BufferReadable& buf, Arena* arena) {
+        read_binary(has_value, buf);
+        if (!has()) {
+            return;
+        }
+        std::string memory_buffer;
+        read_binary(memory_buffer, buf);
+        const auto* p =
+                column_type->deserialize(memory_buffer.data(), &column_data, be_exec_version);
+        DCHECK_EQ(p, memory_buffer.data() + memory_buffer.size());
+    }
+
+    void change_if_better(const IColumn& column, size_t row_num, Arena* arena) {
+        this->change_first_time(column, row_num, nullptr);
+    }
+
+    void change_if_better(const Self& to, Arena* arena) { this->change_first_time(to, nullptr); }
+
+private:
+    bool has_value = false;
+    MutableColumnPtr column_data;
+    DataTypePtr column_type;
+    int be_exec_version = -1;
 };
 
 template <typename Data>
@@ -539,6 +628,7 @@ class AggregateFunctionsSingleValue final
 private:
     DataTypePtr& type;
     using Base = IAggregateFunctionDataHelper<Data, AggregateFunctionsSingleValue<Data>>;
+    using IAggregateFunction::argument_types;
 
 public:
     AggregateFunctionsSingleValue(const DataTypes& arguments)
@@ -555,6 +645,14 @@ class AggregateFunctionsSingleValue final
         }
     }
 
+    void create(AggregateDataPtr __restrict place) const override {
+        if constexpr (std::is_same_v<Data, SingleValueDataComplexType>) {
+            new (place) Data(argument_types, IAggregateFunction::version);
+        } else {
+            new (place) Data;
+        }
+    }
+
     String get_name() const override { return Data::name(); }
 
     DataTypePtr get_return_type() const override { return type; }
@@ -716,4 +814,9 @@ AggregateFunctionPtr create_aggregate_function_single_value(const String& name,
                                                             const DataTypes& argument_types,
                                                             const bool result_is_nullable,
                                                             const AggregateFunctionAttr& attr = {});
+
+template <template <typename> class Data>
+AggregateFunctionPtr create_aggregate_function_single_value_any_value_function(
+        const String& name, const DataTypes& argument_types, const bool result_is_nullable,
+        const AggregateFunctionAttr& attr = {});
 } // namespace doris::vectorized

regression-test/data/nereids_syntax_p0/any_value.out

@@ -5,3 +5,77 @@
 -- !sql_min --
 00:00:00
 
+-- !sql_any1 --
+\N	\N	\N	\N	\N	\N	\N	\N
+
+-- !sql_any2 --
+
+-- !sql_any3 --
+[1, 2, 3]	[4, 5, 6]	["a", "b", "c"]	["d", "e", "f"]	{"s_id":1, "s_name":"a", "s_address":"b"}	{"s_id":2, "s_name":"c", "s_address":"d"}	{"a":1, "b":2}	{"c":3, "d":4}
+
+-- !sql_any4 --
+[1, 2, 3]	[4, 5, 6]	["a", "b", "c"]	["d", "e", "f"]	{"s_id":1, "s_name":"a", "s_address":"b"}	{"s_id":2, "s_name":"c", "s_address":"d"}	{"a":1, "b":2}	{"c":3, "d":4}
+[4, 5, 6]	[7, 8, 9]	["d", "e", "f"]	["g", "h", "i"]	{"s_id":3, "s_name":"e", "s_address":"f"}	{"s_id":4, "s_name":"g", "s_address":"h"}	{"e":5, "f":6}	{"g":7, "h":8}
+
+-- !sql_any5 --
+\N	\N	\N	\N	\N	\N	\N	\N	\N	\N	\N
+
+-- !sql_any6 --
+
+-- !sql_any7 --
+1	1	1	1	1.100	a	2021-01-01	2021-01-01T00:00	a	1.1	1.1
+
+-- !sql_any8 --
+1	1	1	1	1.100	a	2021-01-01	2021-01-01T00:00	a	1.1	1.1
+2	2	2	2	2.200	b	2021-02-02	2021-02-02T00:00	b	2.2	2.2
+
+-- !sql_any9 --
+\N
+
+-- !sql_any10 --
+
+-- !sql_any11 --
+1,2,3
+
+-- !sql_any12 --
+20230101	1,2,3
+20230102	4,5,6
+
+-- !sql_any13 --
+0
+
+-- !sql_any14 --
+
+-- !sql_any15 --
+1
+
+-- !sql_any16 --
+1	1
+2	1
+
+-- !sql_any17 --
+0.0
+
+-- !sql_any18 --
+
+-- !sql_any19 --
+1.0
+
+-- !sql_any20 --
+20230101	1.0
+20230102	200.0
+
+-- !sql_any21 --
+\N	\N	\N
+
+-- !sql_any22 --
+
+-- !sql_any23 --
+1	a
+
+-- !sql_any24 --
+2	a,a
+
+-- !sql_any25 --
+{"DB1":168939,"DNT":"2023-06-10 03:55:33"}
+