branch-3.0: [improve](function) support collect_list with nested types param #47965 (#48114)

Cherry-picked from #47965

Co-authored-by: amory <[email protected]>

Author: github-actions[bot]

be/src/vec/aggregate_functions/aggregate_function_collect.cpp

@@ -48,6 +48,11 @@ AggregateFunctionPtr do_create_agg_function_collect(bool distinct, const DataTyp
                     AggregateFunctionCollectListData<T, HasLimit>, HasLimit, std::false_type>>(
                     argument_types, result_is_nullable);
         }
+    } else if (!distinct) {
+        // void type means support array/map/struct type for collect_list
+        return creator_without_type::create<AggregateFunctionCollect<
+                AggregateFunctionCollectListData<void, HasLimit>, HasLimit, std::false_type>>(
+                argument_types, result_is_nullable);
     }
     return nullptr;
 }
@@ -92,6 +97,9 @@ AggregateFunctionPtr create_aggregate_function_collect_impl(const std::string& n
         if constexpr (ShowNull::value) {
             return do_create_agg_function_collect<void, HasLimit, ShowNull>(
                     distinct, argument_types, result_is_nullable);
+        } else {
+            return do_create_agg_function_collect<void, HasLimit, ShowNull>(
+                    distinct, argument_types, result_is_nullable);
         }
     }
 

be/src/vec/aggregate_functions/aggregate_function_collect.h

@@ -193,6 +193,9 @@ struct AggregateFunctionCollectListData {
     PaddedPODArray<ElementType> data;
     Int64 max_size = -1;
 
+    AggregateFunctionCollectListData() {}
+    AggregateFunctionCollectListData(const DataTypes& argument_types) {}
+
     size_t size() const { return data.size(); }
 
     void add(const IColumn& column, size_t row_num) {
@@ -305,6 +308,67 @@ struct AggregateFunctionCollectListData<StringRef, HasLimit> {
     }
 };
 
+template <typename HasLimit>
+struct AggregateFunctionCollectListData<void, HasLimit> {
+    using ElementType = StringRef;
+    using Self = AggregateFunctionCollectListData<void, HasLimit>;
+    MutableColumnPtr column_data;
+    Int64 max_size = -1;
+
+    AggregateFunctionCollectListData() {}
+    AggregateFunctionCollectListData(const DataTypes& argument_types) {
+        DataTypePtr column_type = argument_types[0];
+        column_data = column_type->create_column();
+    }
+
+    size_t size() const { return column_data->size(); }
+
+    void add(const IColumn& column, size_t row_num) { column_data->insert_from(column, row_num); }
+
+    void merge(const AggregateFunctionCollectListData& rhs) {
+        if constexpr (HasLimit::value) {
+            if (max_size == -1) {
+                max_size = rhs.max_size;
+            }
+            max_size = rhs.max_size;
+
+            column_data->insert_range_from(
+                    *rhs.column_data, 0,
+                    std::min(assert_cast<size_t, TypeCheckOnRelease::DISABLE>(
+                                     static_cast<size_t>(max_size - size())),
+                             rhs.size()));
+        } else {
+            column_data->insert_range_from(*rhs.column_data, 0, rhs.size());
+        }
+    }
+
+    void write(BufferWritable& buf) const {
+        const size_t size = column_data->size();
+        write_binary(size, buf);
+        for (size_t i = 0; i < size; i++) {
+            write_string_binary(column_data->get_data_at(i), buf);
+        }
+        write_var_int(max_size, buf);
+    }
+
+    void read(BufferReadable& buf) {
+        size_t size = 0;
+        read_binary(size, buf);
+        column_data->reserve(size);
+
+        StringRef s;
+        for (size_t i = 0; i < size; i++) {
+            read_string_binary(s, buf);
+            column_data->insert_data(s.data, s.size);
+        }
+        read_var_int(max_size, buf);
+    }
+
+    void reset() { column_data->clear(); }
+
+    void insert_result_into(IColumn& to) const { to.insert_range_from(*column_data, 0, size()); }
+};
+
 template <typename T>
 struct AggregateFunctionArrayAggData {
     using ElementType = T;
@@ -622,7 +686,11 @@ class AggregateFunctionCollect
                 new (place) Data();
             }
         } else {
-            new (place) Data();
+            if constexpr (std::is_same_v<Data, AggregateFunctionCollectListData<void, HasLimit>>) {
+                new (place) Data(argument_types);
+            } else {
+                new (place) Data();
+            }
         }
     }