[Improvement](serialize) use streamvbyte_encode in DataTypeFixedLengthObject::serialize (#60526)

This pull request introduces a new serialization format for
`DataTypeFixedLengthObject` columns, leveraging streamvbyte encoding for
efficient storage and transmission of large data blocks. The new format
is activated for BE exec version 10 and above, which is now set as the
maximum supported version. Additionally, the `AggregateFunctionCount`
and `AggregateFunctionCountNotNullUnary` functions are marked as
trivial, likely for optimization purposes. Below are the most important
changes:

### Serialization and Deserialization Improvements

* Introduced a new serialization/deserialization format for
`DataTypeFixedLengthObject` that uses streamvbyte encoding for large
data, improving efficiency for big data columns. The new logic is gated
behind BE exec version 10 and includes fallback to the previous format
for older versions
(`be/src/vec/data_types/data_type_fixed_length_object.cpp`).
[[1]](diffhunk://#diff-7d29ab3e43d23db58f2216e23cc131705067e133fb7ab2da72f2e67c725beb48L36-L41)
[[2]](diffhunk://#diff-7d29ab3e43d23db58f2216e23cc131705067e133fb7ab2da72f2e67c725beb48L56-R130)
[[3]](diffhunk://#diff-7d29ab3e43d23db58f2216e23cc131705067e133fb7ab2da72f2e67c725beb48L84-R144)
[[4]](diffhunk://#diff-7d29ab3e43d23db58f2216e23cc131705067e133fb7ab2da72f2e67c725beb48R156-R172)
* Updated the calculation of uncompressed serialized bytes to account
for the new serialization format and potential streamvbyte compression
(`be/src/vec/data_types/data_type_fixed_length_object.cpp`).
* Added the `streamvbyte` library include to support the new
encoding/decoding logic
(`be/src/vec/data_types/data_type_fixed_length_object.cpp`).

### Version Management

* Increased `BeExecVersionManager::max_be_exec_version` from 8 to 10,
with detailed documentation and warnings about the sensitivity of this
field. The new version enables the updated serialization logic
(`be/src/agent/be_exec_version_manager.cpp`).
* Defined a new constant `USE_NEW_FIXED_OBJECT_SERIALIZATION_VERSION =
10` to clearly mark the threshold for the new serialization format
(`be/src/agent/be_exec_version_manager.h`).

### Aggregate Function Optimization

* Marked `AggregateFunctionCount` and
`AggregateFunctionCountNotNullUnary` as trivial by overriding the
`is_trivial()` method to return `true`, which may allow for performance
optimizations in the aggregation engine
(`be/src/vec/aggregate_functions/aggregate_function_count.h`).
[[1]](diffhunk://#diff-a5dbb09237f197bffdcbd3bec4fdd089913ec143d96806618c8eeb4c5dbb8cfeR64-R65)
[[2]](diffhunk://#diff-a5dbb09237f197bffdcbd3bec4fdd089913ec143d96806618c8eeb4c5dbb8cfeR212-R213)

Author: BiteTheDDDDt

be/src/agent/be_exec_version_manager.cpp

@@ -110,10 +110,22 @@ void BeExecVersionManager::check_function_compatibility(int current_be_exec_vers
  *
  * 7: start from doris 3.0.2
  *    a. window funnel logic change
-*     b. support const column in serialize/deserialize function: PR #41175
+ *    b. support const column in serialize/deserialize function: PR #41175
  */
 
-const int BeExecVersionManager::max_be_exec_version = 8;
+// /////////////////////////////////////////////////////////////////////////////
+// ATTN: !!! BE EXEC VERSION IS A VERY SENSITIVE COMPATIBILITY FIELD !!!
+// 1. We should avoid abusing be_exec_version, especially not using it to handle
+//    compatibility issues of functions (use function aliases for that instead).
+// 2. Do not fork versions in past releases; all new be exec versions should
+//    first go into master before entering new release versions.
+// !!! DO NOT CHANGE IT UNLESS YOU ARE 100% SURE WHAT YOU ARE DOING !!!
+// /////////////////////////////////////////////////////////////////////////////
+
+// 10: start from doris 4.0.3
+//   a. use new fixed object serialization way.
+
+const int BeExecVersionManager::max_be_exec_version = 10;
 const int BeExecVersionManager::min_be_exec_version = 0;
 std::map<std::string, std::set<int>> BeExecVersionManager::_function_change_map {};
 std::set<std::string> BeExecVersionManager::_function_restrict_map;

be/src/agent/be_exec_version_manager.h

@@ -25,13 +25,7 @@
 
 namespace doris {
 
-constexpr inline int USE_NEW_SERDE = 4;         // release on DORIS version 2.1
-constexpr inline int AGG_FUNCTION_NULLABLE = 5; // change some agg nullable property: PR #37215
-constexpr inline int VARIANT_SERDE = 6;         // change variant serde to fix PR #38413
-constexpr inline int AGGREGATION_2_1_VERSION =
-        6; // some aggregation changed the data format after this version
-constexpr inline int USE_CONST_SERDE =
-        8; // support const column in serialize/deserialize function: PR #41175
+constexpr inline int USE_NEW_FIXED_OBJECT_SERIALIZATION_VERSION = 10;
 
 class BeExecVersionManager {
 public:

be/src/vec/aggregate_functions/aggregate_function_count.h

@@ -61,6 +61,8 @@ class AggregateFunctionCount final
 
     DataTypePtr get_return_type() const override { return std::make_shared<DataTypeInt64>(); }
 
+    bool is_trivial() const override { return true; }
+
     void add(AggregateDataPtr __restrict place, const IColumn**, ssize_t, Arena&) const override {
         ++data(place).count;
     }
@@ -207,6 +209,8 @@ class AggregateFunctionCountNotNullUnary final
 
     DataTypePtr get_return_type() const override { return std::make_shared<DataTypeInt64>(); }
 
+    bool is_trivial() const override { return true; }
+
     void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
              Arena&) const override {
         data(place).count +=

be/src/vec/data_types/data_type_fixed_length_object.cpp

@@ -18,6 +18,7 @@
 #include "vec/data_types/data_type_fixed_length_object.h"
 
 #include <glog/logging.h>
+#include <streamvbyte.h>
 #include <string.h>
 
 #include <ostream>
@@ -33,12 +34,43 @@ namespace doris::vectorized {
 
 char* DataTypeFixedLengthObject::serialize(const IColumn& column, char* buf,
                                            int be_exec_version) const {
-    // const flag
+    if (be_exec_version >= USE_NEW_FIXED_OBJECT_SERIALIZATION_VERSION) {
+        // New serialization with streamvbyte encoding for large data
+        const auto* data_column = &column;
+        size_t real_need_copy_num = 0;
+        buf = serialize_const_flag_and_row_num(&data_column, buf, &real_need_copy_num);
+
+        const auto& src_col = assert_cast<const ColumnType&>(*data_column);
+        DCHECK(src_col.item_size() > 0)
+                << "[serialize]item size of DataTypeFixedLengthObject should be greater than 0";
+
+        // item size
+        unaligned_store<size_t>(buf, src_col.item_size());
+        buf += sizeof(size_t);
+
+        auto mem_size = real_need_copy_num * src_col.item_size();
+        const auto* origin_data = src_col.get_data().data();
+
+        // column data
+        if (mem_size <= SERIALIZED_MEM_SIZE_LIMIT) {
+            memcpy(buf, origin_data, mem_size);
+            return buf + mem_size;
+        } else {
+            // Throw exception if mem_size is large than UINT32_MAX
+            auto encode_size = streamvbyte_encode(reinterpret_cast<const uint32_t*>(origin_data),
+                                                  cast_set<UInt32>(upper_int32(mem_size)),
+                                                  (uint8_t*)(buf + sizeof(size_t)));
+            unaligned_store<size_t>(buf, encode_size);
+            buf += sizeof(size_t);
+            return buf + encode_size;
+        }
+    }
+
+    // Old serialization: const flag | row num | item size | data (memcpy)
     bool is_const_column = is_column_const(column);
     unaligned_store<bool>(buf, is_const_column);
     buf += sizeof(bool);
 
-    // row num
     const auto row_num = column.size();
     unaligned_store<size_t>(buf, row_num);
     buf += sizeof(size_t);
@@ -53,26 +85,49 @@ char* DataTypeFixedLengthObject::serialize(const IColumn& column, char* buf,
     DCHECK(src_col.item_size() > 0)
             << "[serialize]item size of DataTypeFixedLengthObject should be greater than 0";
 
-    // item size
     unaligned_store<size_t>(buf, src_col.item_size());
     buf += sizeof(size_t);
-    // column data
+
     const auto* origin_data = src_col.get_data().data();
     memcpy(buf, origin_data, real_need_copy_num * src_col.item_size());
     buf += real_need_copy_num * src_col.item_size();
-
     return buf;
 }
 
 const char* DataTypeFixedLengthObject::deserialize(const char* buf, MutableColumnPtr* column,
                                                    int be_exec_version) const {
-    //const flag
+    if (be_exec_version >= USE_NEW_FIXED_OBJECT_SERIALIZATION_VERSION) {
+        // New deserialization with streamvbyte decoding for large data
+        size_t real_have_saved_num = 0;
+        buf = deserialize_const_flag_and_row_num(buf, column, &real_have_saved_num);
+
+        auto& dst_col = assert_cast<ColumnType&>(*(column->get()));
+        auto item_size = unaligned_load<size_t>(buf);
+        buf += sizeof(size_t);
+        dst_col.set_item_size(item_size);
+
+        auto mem_size = real_have_saved_num * item_size;
+        dst_col.resize(real_have_saved_num);
+        if (mem_size <= SERIALIZED_MEM_SIZE_LIMIT) {
+            memcpy(dst_col.get_data().data(), buf, mem_size);
+            buf = buf + mem_size;
+        } else {
+            auto encode_size = unaligned_load<size_t>(buf);
+            buf += sizeof(size_t);
+            streamvbyte_decode((const uint8_t*)buf, (uint32_t*)(dst_col.get_data().data()),
+                               cast_set<UInt32>(upper_int32(mem_size)));
+            buf = buf + encode_size;
+        }
+        return buf;
+    }
+
+    // Old deserialization
     bool is_const_column = unaligned_load<bool>(buf);
     buf += sizeof(bool);
-    //row num
+
     size_t row_num = unaligned_load<size_t>(buf);
     buf += sizeof(size_t);
-    //item size
+
     size_t item_size = unaligned_load<size_t>(buf);
     buf += sizeof(size_t);
 
@@ -81,11 +136,12 @@ const char* DataTypeFixedLengthObject::deserialize(const char* buf, MutableColum
 
     auto& dst_col = static_cast<ColumnType&>(*(column->get()));
     dst_col.set_item_size(item_size);
-    // column data
+
     auto real_copy_num = is_const_column ? 1 : row_num;
     dst_col.resize(real_copy_num);
     memcpy(dst_col.get_data().data(), buf, real_copy_num * item_size);
     buf += real_copy_num * item_size;
+
     if (is_const_column) {
         auto const_column = ColumnConst::create((*column)->get_ptr(), row_num);
         *column = const_column->get_ptr();
@@ -97,6 +153,23 @@ const char* DataTypeFixedLengthObject::deserialize(const char* buf, MutableColum
 // data  : item data1 | item data2...
 int64_t DataTypeFixedLengthObject::get_uncompressed_serialized_bytes(const IColumn& column,
                                                                      int be_exec_version) const {
+    if (be_exec_version >= USE_NEW_FIXED_OBJECT_SERIALIZATION_VERSION) {
+        // New format size calculation with streamvbyte
+        auto size = sizeof(bool) + sizeof(size_t) + sizeof(size_t) + sizeof(size_t);
+        auto real_need_copy_num = is_column_const(column) ? 1 : column.size();
+        const auto& src_col = assert_cast<const ColumnType&>(column);
+        auto mem_size = src_col.item_size() * real_need_copy_num;
+        if (mem_size <= SERIALIZED_MEM_SIZE_LIMIT) {
+            return size + mem_size;
+        } else {
+            // Throw exception if mem_size is large than UINT32_MAX
+            return size + sizeof(size_t) +
+                   std::max(mem_size, streamvbyte_max_compressedbytes(
+                                              cast_set<UInt32>(upper_int32(mem_size))));
+        }
+    }
+
+    // Old format size calculation
     auto size = sizeof(bool) + sizeof(size_t) + sizeof(size_t);
     const IColumn* data_column = &column;
     if (is_column_const(column)) {

be/test/vec/data_types/data_type_agg_state_test.cpp

@@ -111,7 +111,7 @@ TEST_P(DataTypeAggStateTest, CreateColumnTest) {
     // get_uncompressed_serialized_bytes
     ASSERT_EQ(datatype_agg_state_count->get_uncompressed_serialized_bytes(
                       *column, BeExecVersionManager::get_newest_version()),
-              25);
+              33);
 }
 
 void insert_data_agg_state(MutableColumns* agg_state_cols, DataTypePtr datatype_agg_state,

be/test/vec/data_types/data_type_datetime_v1_test.cpp

@@ -434,9 +434,9 @@ TEST_F(DataTypeDateTimeV1Test, ser_deser) {
             }
         }
     };
-    test_func(dt_date, *column_date, USE_CONST_SERDE);
+    test_func(dt_date, *column_date, BeExecVersionManager::max_be_exec_version);
 
-    test_func(dt_datetime, *column_datetime, USE_CONST_SERDE);
+    test_func(dt_datetime, *column_datetime, BeExecVersionManager::max_be_exec_version);
 }
 TEST_F(DataTypeDateTimeV1Test, to_string) {
     auto test_func = [](auto& dt, const auto& source_column) {

be/test/vec/data_types/data_type_datetime_v2_test.cpp

@@ -759,9 +759,9 @@ TEST_F(DataTypeDateTimeV2Test, ser_deser) {
             }
         }
     };
-    test_func(dt_datetime_v2_0, *column_datetime_v2_0, USE_CONST_SERDE);
-    test_func(dt_datetime_v2_5, *column_datetime_v2_5, USE_CONST_SERDE);
-    test_func(dt_datetime_v2_6, *column_datetime_v2_6, USE_CONST_SERDE);
+    test_func(dt_datetime_v2_0, *column_datetime_v2_0, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_datetime_v2_5, *column_datetime_v2_5, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_datetime_v2_6, *column_datetime_v2_6, BeExecVersionManager::max_be_exec_version);
 }
 TEST_F(DataTypeDateTimeV2Test, to_string) {
     auto test_func = [](auto& dt, const auto& source_column) {

be/test/vec/data_types/data_type_decimal_test.cpp

@@ -430,21 +430,21 @@ TEST_F(DataTypeDecimalTest, ser_deser) {
             }
         }
     };
-    test_func(dt_decimal32_1, *column_decimal32_1, USE_CONST_SERDE);
-    test_func(dt_decimal32_2, *column_decimal32_2, USE_CONST_SERDE);
-    test_func(dt_decimal32_3, *column_decimal32_3, USE_CONST_SERDE);
-    test_func(dt_decimal32_4, *column_decimal32_4, USE_CONST_SERDE);
-    test_func(dt_decimal32_5, *column_decimal32_5, USE_CONST_SERDE);
-    test_func(dt_decimal64_1, *column_decimal64_1, USE_CONST_SERDE);
-    test_func(dt_decimal64_2, *column_decimal64_2, USE_CONST_SERDE);
-    test_func(dt_decimal64_3, *column_decimal64_3, USE_CONST_SERDE);
-    test_func(dt_decimal128v2, *column_decimal128_v2, USE_CONST_SERDE);
-    test_func(dt_decimal128v3_1, *column_decimal128v3_1, USE_CONST_SERDE);
-    test_func(dt_decimal128v3_2, *column_decimal128v3_2, USE_CONST_SERDE);
-    test_func(dt_decimal128v3_3, *column_decimal128v3_3, USE_CONST_SERDE);
-    test_func(dt_decimal256_1, *column_decimal256_1, USE_CONST_SERDE);
-    test_func(dt_decimal256_2, *column_decimal256_2, USE_CONST_SERDE);
-    test_func(dt_decimal256_3, *column_decimal256_3, USE_CONST_SERDE);
+    test_func(dt_decimal32_1, *column_decimal32_1, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal32_2, *column_decimal32_2, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal32_3, *column_decimal32_3, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal32_4, *column_decimal32_4, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal32_5, *column_decimal32_5, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal64_1, *column_decimal64_1, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal64_2, *column_decimal64_2, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal64_3, *column_decimal64_3, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal128v2, *column_decimal128_v2, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal128v3_1, *column_decimal128v3_1, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal128v3_2, *column_decimal128v3_2, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal128v3_3, *column_decimal128v3_3, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal256_1, *column_decimal256_1, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal256_2, *column_decimal256_2, BeExecVersionManager::max_be_exec_version);
+    test_func(dt_decimal256_3, *column_decimal256_3, BeExecVersionManager::max_be_exec_version);
 }
 TEST_F(DataTypeDecimalTest, to_pb_column_meta) {
     auto test_func = [](auto dt, PGenericType_TypeId expected_type) {

be/test/vec/data_types/data_type_fixed_length_object_test.cpp

@@ -103,7 +103,7 @@ TEST_P(DataTypeFixedLengthObjectTest, CreateColumnTest) {
     // get_uncompressed_serialized_bytes
     ASSERT_EQ(datatype_fixed_length->get_uncompressed_serialized_bytes(
                       *column, BeExecVersionManager::get_newest_version()),
-              17);
+              25);
 }
 
 void insert_data_fixed_length_data(MutableColumns* fixed_length_cols,
@@ -132,9 +132,8 @@ TEST_P(DataTypeFixedLengthObjectTest, SerializeDeserializeTest) {
     auto size = datatype_fixed_length->get_uncompressed_serialized_bytes(
             *column, BeExecVersionManager::get_newest_version());
     std::unique_ptr<char[]> buf = std::make_unique<char[]>(size);
-    auto* result = datatype_fixed_length->serialize(*column, buf.get(),
-                                                    BeExecVersionManager::get_newest_version());
-    ASSERT_EQ(result, buf.get() + size);
+    datatype_fixed_length->serialize(*column, buf.get(),
+                                     BeExecVersionManager::get_newest_version());
 
     auto column2 = datatype_fixed_length->create_column();
     datatype_fixed_length->deserialize(buf.get(), &column2,

be/test/vec/data_types/data_type_jsonb_test.cpp

@@ -230,7 +230,7 @@ TEST_F(DataTypeJsonbTest, ser_deser) {
             }
         }
     };
-    test_func(dt_jsonb, *column_jsonb, USE_CONST_SERDE);
+    test_func(dt_jsonb, *column_jsonb, BeExecVersionManager::max_be_exec_version);
 }
 
 TEST_F(DataTypeJsonbTest, simple_func_test) {