Merge pull request ClickHouse#91310 from ClickHouse/backport/25.8/90880

Backport ClickHouse#90880 to 25.8: Add setting for compatibility serialziation of string

Author: antonio2368

src/AggregateFunctions/AggregateFunctionGroupArray.cpp

@@ -457,7 +457,8 @@ struct GroupArrayNodeGeneral : public GroupArrayNodeBase<GroupArrayNodeGeneral>
     static Node * allocate(const IColumn & column, size_t row_num, Arena * arena)
     {
         const char * begin = arena->alignedAlloc(sizeof(Node), alignof(Node));
-        StringRef value = column.serializeAggregationStateValueIntoArena(row_num, *arena, begin);
+        auto settings = IColumn::SerializationSettings::createForAggregationState();
+        auto value = column.serializeValueIntoArena(row_num, *arena, begin, &settings);
 
         Node * node = reinterpret_cast<Node *>(const_cast<char *>(begin));
         node->size = value.size;

src/AggregateFunctions/AggregateFunctionGroupArrayIntersect.cpp

@@ -236,7 +236,8 @@ class AggregateFunctionGroupArrayIntersectGeneric final
                 else
                 {
                     const char * begin = nullptr;
-                    StringRef serialized = data_column->serializeAggregationStateValueIntoArena(offset + i, *arena, begin);
+                    auto settings = IColumn::SerializationSettings::createForAggregationState();
+                    auto serialized = data_column->serializeValueIntoArena(offset + i, *arena, begin, &settings);
                     chassert(serialized.data != nullptr);
                     set.emplace(SerializedKeyHolder{serialized, *arena}, it, inserted);
                 }
@@ -256,7 +257,8 @@ class AggregateFunctionGroupArrayIntersectGeneric final
                 else
                 {
                     const char * begin = nullptr;
-                    StringRef serialized = data_column->serializeAggregationStateValueIntoArena(offset + i, *arena, begin);
+                    auto settings = IColumn::SerializationSettings::createForAggregationState();
+                    auto serialized = data_column->serializeValueIntoArena(offset + i, *arena, begin, &settings);
                     chassert(serialized.data != nullptr);
                     it = set.find(serialized);
 

src/AggregateFunctions/AggregateFunctionTopK.cpp

@@ -310,7 +310,8 @@ class AggregateFunctionTopKGeneric final
         else
         {
             const char * begin = nullptr;
-            StringRef str_serialized = columns[0]->serializeAggregationStateValueIntoArena(row_num, *arena, begin);
+            auto settings = IColumn::SerializationSettings::createForAggregationState();
+            auto str_serialized = columns[0]->serializeValueIntoArena(row_num, *arena, begin, &settings);
             if constexpr (is_weighted)
                 set.insert(str_serialized, columns[1]->getUInt(row_num));
             else

src/AggregateFunctions/Combinators/AggregateFunctionDistinct.h

@@ -116,7 +116,8 @@ struct AggregateFunctionDistinctMultipleGenericData : public AggregateFunctionDi
         StringRef value(begin, 0);
         for (size_t i = 0; i < columns_num; ++i)
         {
-            auto cur_ref = columns[i]->serializeAggregationStateValueIntoArena(row_num, *arena, begin);
+            auto settings = IColumn::SerializationSettings::createForAggregationState();
+            auto cur_ref = columns[i]->serializeValueIntoArena(row_num, *arena, begin, &settings);
             value.data = cur_ref.data - value.size;
             value.size += cur_ref.size;
         }
@@ -140,8 +141,9 @@ struct AggregateFunctionDistinctMultipleGenericData : public AggregateFunctionDi
                 history.emplace(ArenaKeyHolder{value, *arena}, it, inserted);
                 ReadBufferFromString in({it->getValue().data, it->getValue().size});
                 /// Multiple columns are serialized one by one
+                auto settings = IColumn::SerializationSettings::createForAggregationState();
                 for (auto & column : argument_columns)
-                    column->deserializeAndInsertAggregationStateValueFromArena(in);
+                    column->deserializeAndInsertFromArena(in, &settings);
             }
         }
     }

src/AggregateFunctions/KeyHolderHelpers.h

@@ -23,7 +23,8 @@ static auto getKeyHolder(const IColumn & column, size_t row_num, Arena & arena)
     else
     {
         const char * begin = nullptr;
-        StringRef serialized = column.serializeAggregationStateValueIntoArena(row_num, arena, begin);
+        auto settings = IColumn::SerializationSettings::createForAggregationState();
+        auto serialized = column.serializeValueIntoArena(row_num, arena, begin, &settings);
         assert(serialized.data != nullptr);
         return SerializedKeyHolder{serialized, arena};
     }
@@ -37,7 +38,8 @@ static void deserializeAndInsert(StringRef str, IColumn & data_to)
     else
     {
         ReadBufferFromString in({str.data, str.size});
-        data_to.deserializeAndInsertAggregationStateValueFromArena(in);
+        auto settings = IColumn::SerializationSettings::createForAggregationState();
+        data_to.deserializeAndInsertFromArena(in, &settings);
         if (!in.eof())
         {
             throw Exception(ErrorCodes::INCORRECT_DATA, "Extra bytes ({}) found after deserializing aggregation state", in.available());

src/Columns/ColumnAggregateFunction.cpp

@@ -601,15 +601,16 @@ void ColumnAggregateFunction::insertDefault()
     pushBackAndCreateState(data, arena, func.get());
 }
 
-StringRef ColumnAggregateFunction::serializeValueIntoArena(size_t n, Arena & arena, const char *& begin) const
+StringRef ColumnAggregateFunction::serializeValueIntoArena(
+    size_t n, Arena & arena, const char *& begin, const IColumn::SerializationSettings *) const
 {
     WriteBufferFromArena out(arena, begin);
     func->serialize(data[n], out, version);
     out.finalize();
     return out.complete();
 }
 
-void ColumnAggregateFunction::deserializeAndInsertFromArena(ReadBuffer & in)
+void ColumnAggregateFunction::deserializeAndInsertFromArena(ReadBuffer & in, const IColumn::SerializationSettings *)
 {
     ensureOwnership();
 

src/Columns/ColumnAggregateFunction.h

@@ -171,9 +171,9 @@ class ColumnAggregateFunction final : public COWHelper<IColumnHelper<ColumnAggre
 
     void insertDefault() override;
 
-    StringRef serializeValueIntoArena(size_t n, Arena & arena, char const *& begin) const override;
+    StringRef serializeValueIntoArena(size_t n, Arena & arena, char const *& begin, const IColumn::SerializationSettings * settings) const override;
 
-    void deserializeAndInsertFromArena(ReadBuffer & in) override;
+    void deserializeAndInsertFromArena(ReadBuffer & in, const IColumn::SerializationSettings * settings) override;
 
     void skipSerializedInArena(ReadBuffer & in) const override;
 

src/Columns/ColumnArray.cpp

@@ -223,7 +223,8 @@ void ColumnArray::insertData(const char * pos, size_t length)
 }
 
 
-StringRef ColumnArray::serializeValueIntoArena(size_t n, Arena & arena, char const *& begin) const
+StringRef
+ColumnArray::serializeValueIntoArena(size_t n, Arena & arena, char const *& begin, const IColumn::SerializationSettings * settings) const
 {
     size_t array_size = sizeAt(n);
     size_t offset = offsetAt(n);
@@ -235,49 +236,27 @@ StringRef ColumnArray::serializeValueIntoArena(size_t n, Arena & arena, char con
 
     for (size_t i = 0; i < array_size; ++i)
     {
-        auto value_ref = getData().serializeValueIntoArena(offset + i, arena, begin);
+        auto value_ref = getData().serializeValueIntoArena(offset + i, arena, begin, settings);
         res.data = value_ref.data - res.size;
         res.size += value_ref.size;
     }
 
     return res;
 }
 
-
-StringRef ColumnArray::serializeAggregationStateValueIntoArena(size_t n, Arena & arena, char const *& begin) const
-{
-    size_t array_size = sizeAt(n);
-    size_t offset = offsetAt(n);
-
-    char * pos = arena.allocContinue(sizeof(array_size), begin);
-    memcpy(pos, &array_size, sizeof(array_size));
-
-    StringRef res(pos, sizeof(array_size));
-
-    for (size_t i = 0; i < array_size; ++i)
-    {
-        auto value_ref = getData().serializeAggregationStateValueIntoArena(offset + i, arena, begin);
-        res.data = value_ref.data - res.size;
-        res.size += value_ref.size;
-    }
-
-    return res;
-}
-
-
-char * ColumnArray::serializeValueIntoMemory(size_t n, char * memory) const
+char * ColumnArray::serializeValueIntoMemory(size_t n, char * memory, const IColumn::SerializationSettings * settings) const
 {
     size_t array_size = sizeAt(n);
     size_t offset = offsetAt(n);
 
     memcpy(memory, &array_size, sizeof(array_size));
     memory += sizeof(array_size);
     for (size_t i = 0; i < array_size; ++i)
-        memory = getData().serializeValueIntoMemory(offset + i, memory);
+        memory = getData().serializeValueIntoMemory(offset + i, memory, settings);
     return memory;
 }
 
-std::optional<size_t> ColumnArray::getSerializedValueSize(size_t n) const
+std::optional<size_t> ColumnArray::getSerializedValueSize(size_t n, const IColumn::SerializationSettings * settings) const
 {
     const auto & offsets_data = getOffsets();
 
@@ -287,7 +266,7 @@ std::optional<size_t> ColumnArray::getSerializedValueSize(size_t n) const
     size_t res = sizeof(offsets_data[0]);
     for (; pos < end; ++pos)
     {
-        auto element_size = getData().getSerializedValueSize(pos);
+        auto element_size = getData().getSerializedValueSize(pos, settings);
         if (!element_size)
             return std::nullopt;
         res += *element_size;
@@ -297,24 +276,13 @@ std::optional<size_t> ColumnArray::getSerializedValueSize(size_t n) const
 }
 
 
-void ColumnArray::deserializeAndInsertFromArena(ReadBuffer & in)
-{
-    size_t array_size;
-    readBinaryLittleEndian<size_t>(array_size, in);
-
-    for (size_t i = 0; i < array_size; ++i)
-        getData().deserializeAndInsertFromArena(in);
-
-    getOffsets().push_back(getOffsets().back() + array_size);
-}
-
-void ColumnArray::deserializeAndInsertAggregationStateValueFromArena(ReadBuffer & in)
+void ColumnArray::deserializeAndInsertFromArena(ReadBuffer & in, const IColumn::SerializationSettings * settings)
 {
     size_t array_size;
     readBinaryLittleEndian<size_t>(array_size, in);
 
     for (size_t i = 0; i < array_size; ++i)
-        getData().deserializeAndInsertAggregationStateValueFromArena(in);
+        getData().deserializeAndInsertFromArena(in, settings);
 
     getOffsets().push_back(getOffsets().back() + array_size);
 }

src/Columns/ColumnArray.h

@@ -78,12 +78,10 @@ class ColumnArray final : public COWHelper<IColumnHelper<ColumnArray>, ColumnArr
     StringRef getDataAt(size_t n) const override;
     bool isDefaultAt(size_t n) const override;
     void insertData(const char * pos, size_t length) override;
-    StringRef serializeValueIntoArena(size_t n, Arena & arena, char const *& begin) const override;
-    StringRef serializeAggregationStateValueIntoArena(size_t n, Arena & arena, char const *& begin) const override;
-    char * serializeValueIntoMemory(size_t, char * memory) const override;
-    std::optional<size_t> getSerializedValueSize(size_t n) const override;
-    void deserializeAndInsertFromArena(ReadBuffer & in) override;
-    void deserializeAndInsertAggregationStateValueFromArena(ReadBuffer & in) override;
+    StringRef serializeValueIntoArena(size_t n, Arena & arena, char const *& begin, const IColumn::SerializationSettings * settings) const override;
+    char * serializeValueIntoMemory(size_t, char * memory, const IColumn::SerializationSettings * settings) const override;
+    std::optional<size_t> getSerializedValueSize(size_t n, const IColumn::SerializationSettings * settings) const override;
+    void deserializeAndInsertFromArena(ReadBuffer & in, const IColumn::SerializationSettings * settings) override;
     void skipSerializedInArena(ReadBuffer & in) const override;
     void updateHashWithValue(size_t n, SipHash & hash) const override;
     WeakHash32 getWeakHash32() const override;

src/Columns/ColumnBLOB.h

@@ -161,9 +161,9 @@ class ColumnBLOB : public COWHelper<IColumnHelper<ColumnBLOB>, ColumnBLOB>
     void insertData(const char *, size_t) override { throwInapplicable(); }
     void insertDefault() override { throwInapplicable(); }
     void popBack(size_t) override { throwInapplicable(); }
-    StringRef serializeValueIntoArena(size_t, Arena &, char const *&) const override { throwInapplicable(); }
-    char * serializeValueIntoMemory(size_t, char *) const override { throwInapplicable(); }
-    void deserializeAndInsertFromArena(ReadBuffer &) override { throwInapplicable(); }
+    StringRef serializeValueIntoArena(size_t, Arena &, char const *&, const IColumn::SerializationSettings *) const override { throwInapplicable(); }
+    char * serializeValueIntoMemory(size_t, char *, const IColumn::SerializationSettings *) const override { throwInapplicable(); }
+    void deserializeAndInsertFromArena(ReadBuffer &, const IColumn::SerializationSettings *) override { throwInapplicable(); }
     void skipSerializedInArena(ReadBuffer &) const override { throwInapplicable(); }
     void updateHashWithValue(size_t, SipHash &) const override { throwInapplicable(); }
     WeakHash32 getWeakHash32() const override { throwInapplicable(); }