branch-3.1: [fix](collect_list) fix collect_list in multi BE will ser/de failed apache#48314 (apache#56752)

picked from apache#48314

Author: amory

be/src/vec/aggregate_functions/aggregate_function_collect.h

@@ -312,13 +312,15 @@ template <typename HasLimit>
 struct AggregateFunctionCollectListData<void, HasLimit> {
     using ElementType = StringRef;
     using Self = AggregateFunctionCollectListData<void, HasLimit>;
+    DataTypeSerDeSPtr serde; // for complex serialize && deserialize from multi BE
     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();
+        serde = column_type->get_serde();
     }
 
     size_t size() const { return column_data->size(); }
@@ -345,21 +347,41 @@ struct AggregateFunctionCollectListData<void, HasLimit> {
     void write(BufferWritable& buf) const {
         const size_t size = column_data->size();
         write_binary(size, buf);
+
+        DataTypeSerDe::FormatOptions opt;
+        auto tmp_str = ColumnString::create();
+        VectorBufferWriter tmp_buf(*tmp_str.get());
+
         for (size_t i = 0; i < size; i++) {
-            write_string_binary(column_data->get_data_at(i), buf);
+            tmp_str->clear();
+            if (Status st = serde->serialize_one_cell_to_json(*column_data, i, tmp_buf, opt); !st) {
+                throw doris::Exception(ErrorCode::INTERNAL_ERROR,
+                                       "Failed to serialize data for " + column_data->get_name() +
+                                               " error: " + st.to_string());
+            }
+            tmp_buf.commit();
+            write_string_binary(tmp_str->get_data_at(0), buf);
         }
+
         write_var_int(max_size, buf);
     }
 
     void read(BufferReadable& buf) {
         size_t size = 0;
         read_binary(size, buf);
+        column_data->clear();
         column_data->reserve(size);
 
         StringRef s;
+        DataTypeSerDe::FormatOptions opt;
         for (size_t i = 0; i < size; i++) {
             read_string_binary(s, buf);
-            column_data->insert_data(s.data, s.size);
+            Slice slice(s.data, s.size);
+            if (Status st = serde->deserialize_one_cell_from_json(*column_data, slice, opt); !st) {
+                throw doris::Exception(ErrorCode::INTERNAL_ERROR,
+                                       "Failed to deserialize data for " + column_data->get_name() +
+                                               " error: " + st.to_string());
+            }
         }
         read_var_int(max_size, buf);
     }
@@ -613,7 +635,6 @@ struct AggregateFunctionArrayAggData<void> {
         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);

be/test/vec/aggregate_functions/agg_collect_test.cpp

@@ -63,12 +63,29 @@ class VAggCollectTest : public testing::Test {
 
     template <typename DataType>
     void agg_collect_add_elements(AggregateFunctionPtr agg_function, AggregateDataPtr place,
-                                  size_t input_nums) {
+                                  size_t input_nums, bool support_complex = false) {
         using FieldType = typename DataType::FieldType;
-        auto type = std::make_shared<DataType>();
-        auto input_col = type->create_column();
+        MutableColumnPtr input_col;
+        if (support_complex) {
+            auto type =
+                    std::make_shared<DataTypeArray>(make_nullable(std::make_shared<DataType>()));
+            input_col = type->create_column();
+        } else {
+            auto type = std::make_shared<DataType>();
+            input_col = type->create_column();
+        }
         for (size_t i = 0; i < input_nums; ++i) {
             for (size_t j = 0; j < _repeated_times; ++j) {
+                if (support_complex) {
+                    if constexpr (std::is_same_v<DataType, DataTypeString>) {
+                        Array vec1 = {Field(String("item0" + std::to_string(i))),
+                                      Field(String("item1" + std::to_string(i)))};
+                        input_col->insert(vec1);
+                    } else {
+                        input_col->insert_default();
+                    }
+                    continue;
+                }
                 if constexpr (std::is_same_v<DataType, DataTypeString>) {
                     auto item = std::string("item") + std::to_string(i);
                     input_col->insert_data(item.c_str(), item.size());
@@ -87,8 +104,13 @@ class VAggCollectTest : public testing::Test {
     }
 
     template <typename DataType>
-    void test_agg_collect(const std::string& fn_name, size_t input_nums = 0) {
+    void test_agg_collect(const std::string& fn_name, size_t input_nums = 0,
+                          bool support_complex = false) {
         DataTypes data_types = {(DataTypePtr)std::make_shared<DataType>()};
+        if (support_complex) {
+            data_types = {
+                    (DataTypePtr)std::make_shared<DataTypeArray>(make_nullable(data_types[0]))};
+        }
         LOG(INFO) << "test_agg_collect for " << fn_name << "(" << data_types[0]->get_name() << ")";
         AggregateFunctionSimpleFactory factory = AggregateFunctionSimpleFactory::instance();
         auto agg_function = factory.get(fn_name, data_types, false, -1);
@@ -98,7 +120,7 @@ class VAggCollectTest : public testing::Test {
         AggregateDataPtr place = memory.get();
         agg_function->create(place);
 
-        agg_collect_add_elements<DataType>(agg_function, place, input_nums);
+        agg_collect_add_elements<DataType>(agg_function, place, input_nums, support_complex);
 
         ColumnString buf;
         VectorBufferWriter buf_writer(buf);
@@ -111,7 +133,7 @@ class VAggCollectTest : public testing::Test {
         AggregateDataPtr place2 = memory2.get();
         agg_function->create(place2);
 
-        agg_collect_add_elements<DataType>(agg_function, place2, input_nums);
+        agg_collect_add_elements<DataType>(agg_function, place2, input_nums, support_complex);
 
         agg_function->merge(place, place2, &_agg_arena_pool);
         auto column_result = ColumnArray::create(data_types[0]->create_column());
@@ -173,4 +195,15 @@ TEST_F(VAggCollectTest, test_with_data) {
     test_agg_collect<DataTypeString>("collect_set", 5);
 }
 
+TEST_F(VAggCollectTest, test_complex_data_type) {
+    test_agg_collect<DataTypeInt8>("collect_list", 7, true);
+    test_agg_collect<DataTypeInt128>("array_agg", 9, true);
+
+    test_agg_collect<DataTypeDateTime>("collect_list", 5, true);
+    test_agg_collect<DataTypeDateTime>("array_agg", 6, true);
+
+    test_agg_collect<DataTypeString>("collect_list", 10, true);
+    test_agg_collect<DataTypeString>("array_agg", 5, true);
+}
+
 } // namespace doris::vectorized