added arena in AggregateFunctionSingleValue for strings and numbers [#CLICKHOUSE-3503]

dbms/src/AggregateFunctions/AggregateFunctionArgMinMax.h

@@ -50,16 +50,16 @@ public:
         return type_res;
     }
 
-    void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
+    void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena * arena) const override
     {
-        if (this->data(place).value.changeIfBetter(*columns[1], row_num))
-            this->data(place).result.change(*columns[0], row_num);
+        if (this->data(place).value.changeIfBetter(*columns[1], row_num, arena))
+            this->data(place).result.change(*columns[0], row_num, arena);
     }
 
-    void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
+    void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena * arena) const override
     {
-        if (this->data(place).value.changeIfBetter(this->data(rhs).value))
-            this->data(place).result.change(this->data(rhs).result);
+        if (this->data(place).value.changeIfBetter(this->data(rhs).value, arena))
+            this->data(place).result.change(this->data(rhs).result, arena);
     }
 
     void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
@@ -68,10 +68,10 @@ public:
         this->data(place).value.write(buf, *type_val);
     }
 
-    void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
+    void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena * arena) const override
     {
-        this->data(place).result.read(buf, *type_res);
-        this->data(place).value.read(buf, *type_val);
+        this->data(place).result.read(buf, *type_res, arena);
+        this->data(place).value.read(buf, *type_val, arena);
     }
 
     void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override

dbms/src/AggregateFunctions/AggregateFunctionMinMaxAny.h

@@ -24,12 +24,13 @@ namespace DB
 template <typename T>
 struct SingleValueDataFixed
 {
+private:
     using Self = SingleValueDataFixed<T>;
 
     bool has_value = false; /// We need to remember if at least one value has been passed. This is necessary for AggregateFunctionIf.
     T value;
 
-
+public:
     bool has() const
     {
         return has_value;
@@ -50,7 +51,7 @@ struct SingleValueDataFixed
             writeBinary(value, buf);
     }
 
-    void read(ReadBuffer & buf, const IDataType & /*data_type*/)
+    void read(ReadBuffer & buf, const IDataType & /*data_type*/, Arena *)
     {
         readBinary(has_value, buf);
         if (has())
@@ -58,96 +59,96 @@ struct SingleValueDataFixed
     }
 
 
-    void change(const IColumn & column, size_t row_num)
+    void change(const IColumn & column, size_t row_num, Arena *)
     {
         has_value = true;
         value = static_cast<const ColumnVector<T> &>(column).getData()[row_num];
     }
 
     /// Assuming to.has()
-    void change(const Self & to)
+    void change(const Self & to, Arena *)
     {
         has_value = true;
         value = to.value;
     }
 
-    bool changeFirstTime(const IColumn & column, size_t row_num)
+    bool changeFirstTime(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (!has())
         {
-            change(column, row_num);
+            change(column, row_num, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeFirstTime(const Self & to)
+    bool changeFirstTime(const Self & to, Arena * arena)
     {
         if (!has() && to.has())
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeEveryTime(const IColumn & column, size_t row_num)
+    bool changeEveryTime(const IColumn & column, size_t row_num, Arena * arena)
     {
-        change(column, row_num);
+        change(column, row_num, arena);
         return true;
     }
 
-    bool changeEveryTime(const Self & to)
+    bool changeEveryTime(const Self & to, Arena * arena)
     {
         if (to.has())
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfLess(const IColumn & column, size_t row_num)
+    bool changeIfLess(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (!has() || static_cast<const ColumnVector<T> &>(column).getData()[row_num] < value)
         {
-            change(column, row_num);
+            change(column, row_num, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfLess(const Self & to)
+    bool changeIfLess(const Self & to, Arena * arena)
     {
         if (to.has() && (!has() || to.value < value))
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfGreater(const IColumn & column, size_t row_num)
+    bool changeIfGreater(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (!has() || static_cast<const ColumnVector<T> &>(column).getData()[row_num] > value)
         {
-            change(column, row_num);
+            change(column, row_num, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfGreater(const Self & to)
+    bool changeIfGreater(const Self & to, Arena * arena)
     {
         if (to.has() && (!has() || to.value > value))
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
@@ -171,25 +172,21 @@ struct SingleValueDataFixed
   */
 struct __attribute__((__packed__, __aligned__(1))) SingleValueDataString
 {
+private:
     using Self = SingleValueDataString;
 
     Int32 size = -1;    /// -1 indicates that there is no value.
+    Int32 capacity = 0;    /// power of two or zero
+    char * __attribute__((__packed__, __aligned__(1))) large_data;
 
+public:
     static constexpr Int32 AUTOMATIC_STORAGE_SIZE = 64;
-    static constexpr Int32 MAX_SMALL_STRING_SIZE = AUTOMATIC_STORAGE_SIZE - sizeof(size);
+    static constexpr Int32 MAX_SMALL_STRING_SIZE = AUTOMATIC_STORAGE_SIZE - sizeof(size) - sizeof(capacity) - sizeof(large_data);
 
-    union __attribute__((__packed__, __aligned__(1)))
-    {
-        char small_data[MAX_SMALL_STRING_SIZE]; /// Including the terminating zero.
-        char * __attribute__((__packed__, __aligned__(1))) large_data;
-    };
-
-    ~SingleValueDataString()
-    {
-        if (size > MAX_SMALL_STRING_SIZE)
-            free(large_data);
-    }
+private:
+    char small_data[MAX_SMALL_STRING_SIZE]; /// Including the terminating zero.
 
+public:
     bool has() const
     {
         return size >= 0;
@@ -220,7 +217,7 @@ struct __attribute__((__packed__, __aligned__(1))) SingleValueDataString
             buf.write(getData(), size);
     }
 
-    void read(ReadBuffer & buf, const IDataType & /*data_type*/)
+    void read(ReadBuffer & buf, const IDataType & /*data_type*/, Arena * arena)
     {
         Int32 rhs_size;
         readBinary(rhs_size, buf);
@@ -229,8 +226,7 @@ struct __attribute__((__packed__, __aligned__(1))) SingleValueDataString
         {
             if (rhs_size <= MAX_SMALL_STRING_SIZE)
             {
-                if (size > MAX_SMALL_STRING_SIZE)
-                    free(large_data);
+                /// Don't free large_data here.
 
                 size = rhs_size;
 
@@ -239,12 +235,11 @@ struct __attribute__((__packed__, __aligned__(1))) SingleValueDataString
             }
             else
             {
-                if (size < rhs_size)
+                if (capacity < rhs_size)
                 {
-                    if (size > MAX_SMALL_STRING_SIZE)
-                        free(large_data);
-
-                    large_data = reinterpret_cast<char *>(malloc(rhs_size));
+                    capacity = static_cast<UInt32>(roundUpToPowerOfTwoOrZero(rhs_size));
+                    /// Don't free large_data here.
+                    large_data = arena->alloc(capacity);
                 }
 
                 size = rhs_size;
@@ -253,22 +248,19 @@ struct __attribute__((__packed__, __aligned__(1))) SingleValueDataString
         }
         else
         {
-            if (size > MAX_SMALL_STRING_SIZE)
-                free(large_data);
+            /// Don't free large_data here.
             size = rhs_size;
         }
     }
 
     /// Assuming to.has()
-    void changeImpl(StringRef value)
+    void changeImpl(StringRef value, Arena * arena)
     {
         Int32 value_size = value.size;
 
         if (value_size <= MAX_SMALL_STRING_SIZE)
         {
-            if (size > MAX_SMALL_STRING_SIZE)
-                free(large_data);
-
+            /// Don't free large_data here.
             size = value_size;
 
             if (size > 0)
@@ -276,12 +268,11 @@ struct __attribute__((__packed__, __aligned__(1))) SingleValueDataString
         }
         else
         {
-            if (size < value_size)
+            if (capacity < value_size)
             {
-                if (size > MAX_SMALL_STRING_SIZE)
-                    free(large_data);
-
-                large_data = reinterpret_cast<char *>(malloc(value.size));
+                /// Don't free large_data here.
+                capacity = roundUpToPowerOfTwoOrZero(value_size);
+                large_data = arena->alloc(capacity);
             }
 
             size = value_size;
@@ -289,93 +280,93 @@ struct __attribute__((__packed__, __aligned__(1))) SingleValueDataString
         }
     }
 
-    void change(const IColumn & column, size_t row_num)
+    void change(const IColumn & column, size_t row_num, Arena * arena)
     {
-        changeImpl(static_cast<const ColumnString &>(column).getDataAtWithTerminatingZero(row_num));
+        changeImpl(static_cast<const ColumnString &>(column).getDataAtWithTerminatingZero(row_num), arena);
     }
 
-    void change(const Self & to)
+    void change(const Self & to, Arena * arena)
     {
-        changeImpl(to.getStringRef());
+        changeImpl(to.getStringRef(), arena);
     }
 
-    bool changeFirstTime(const IColumn & column, size_t row_num)
+    bool changeFirstTime(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (!has())
         {
-            change(column, row_num);
+            change(column, row_num, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeFirstTime(const Self & to)
+    bool changeFirstTime(const Self & to, Arena * arena)
     {
         if (!has() && to.has())
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeEveryTime(const IColumn & column, size_t row_num)
+    bool changeEveryTime(const IColumn & column, size_t row_num, Arena * arena)
     {
-        change(column, row_num);
+        change(column, row_num, arena);
         return true;
     }
 
-    bool changeEveryTime(const Self & to)
+    bool changeEveryTime(const Self & to, Arena * arena)
     {
         if (to.has())
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfLess(const IColumn & column, size_t row_num)
+    bool changeIfLess(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (!has() || static_cast<const ColumnString &>(column).getDataAtWithTerminatingZero(row_num) < getStringRef())
         {
-            change(column, row_num);
+            change(column, row_num, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfLess(const Self & to)
+    bool changeIfLess(const Self & to, Arena * arena)
     {
         if (to.has() && (!has() || to.getStringRef() < getStringRef()))
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfGreater(const IColumn & column, size_t row_num)
+    bool changeIfGreater(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (!has() || static_cast<const ColumnString &>(column).getDataAtWithTerminatingZero(row_num) > getStringRef())
         {
-            change(column, row_num);
+            change(column, row_num, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfGreater(const Self & to)
+    bool changeIfGreater(const Self & to, Arena * arena)
     {
         if (to.has() && (!has() || to.getStringRef() > getStringRef()))
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
@@ -401,10 +392,12 @@ static_assert(
 /// For any other value types.
 struct SingleValueDataGeneric
 {
+private:
     using Self = SingleValueDataGeneric;
 
     Field value;
 
+public:
     bool has() const
     {
         return !value.isNull();
@@ -429,7 +422,7 @@ struct SingleValueDataGeneric
             writeBinary(false, buf);
     }
 
-    void read(ReadBuffer & buf, const IDataType & data_type)
+    void read(ReadBuffer & buf, const IDataType & data_type, Arena *)
     {
         bool is_not_null;
         readBinary(is_not_null, buf);
@@ -438,60 +431,60 @@ struct SingleValueDataGeneric
             data_type.deserializeBinary(value, buf);
     }
 
-    void change(const IColumn & column, size_t row_num)
+    void change(const IColumn & column, size_t row_num, Arena *)
     {
         column.get(row_num, value);
     }
 
-    void change(const Self & to)
+    void change(const Self & to, Arena *)
     {
         value = to.value;
     }
 
-    bool changeFirstTime(const IColumn & column, size_t row_num)
+    bool changeFirstTime(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (!has())
         {
-            change(column, row_num);
+            change(column, row_num, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeFirstTime(const Self & to)
+    bool changeFirstTime(const Self & to, Arena * arena)
     {
         if (!has() && to.has())
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeEveryTime(const IColumn & column, size_t row_num)
+    bool changeEveryTime(const IColumn & column, size_t row_num, Arena * arena)
     {
-        change(column, row_num);
+        change(column, row_num, arena);
         return true;
     }
 
-    bool changeEveryTime(const Self & to)
+    bool changeEveryTime(const Self & to, Arena * arena)
     {
         if (to.has())
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfLess(const IColumn & column, size_t row_num)
+    bool changeIfLess(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (!has())
         {
-            change(column, row_num);
+            change(column, row_num, arena);
             return true;
         }
         else
@@ -508,22 +501,22 @@ struct SingleValueDataGeneric
         }
     }
 
-    bool changeIfLess(const Self & to)
+    bool changeIfLess(const Self & to, Arena * arena)
     {
         if (to.has() && (!has() || to.value < value))
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
             return false;
     }
 
-    bool changeIfGreater(const IColumn & column, size_t row_num)
+    bool changeIfGreater(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (!has())
         {
-            change(column, row_num);
+            change(column, row_num, arena);
             return true;
         }
         else
@@ -540,11 +533,11 @@ struct SingleValueDataGeneric
         }
     }
 
-    bool changeIfGreater(const Self & to)
+    bool changeIfGreater(const Self & to, Arena * arena)
     {
         if (to.has() && (!has() || to.value > value))
         {
-            change(to);
+            change(to, arena);
             return true;
         }
         else
@@ -573,8 +566,8 @@ struct AggregateFunctionMinData : Data
 {
     using Self = AggregateFunctionMinData<Data>;
 
-    bool changeIfBetter(const IColumn & column, size_t row_num) { return this->changeIfLess(column, row_num); }
-    bool changeIfBetter(const Self & to)                        { return this->changeIfLess(to); }
+    bool changeIfBetter(const IColumn & column, size_t row_num, Arena * arena) { return this->changeIfLess(column, row_num, arena); }
+    bool changeIfBetter(const Self & to, Arena * arena)                        { return this->changeIfLess(to, arena); }
 
     static const char * name() { return "min"; }
 };
@@ -584,8 +577,8 @@ struct AggregateFunctionMaxData : Data
 {
     using Self = AggregateFunctionMaxData<Data>;
 
-    bool changeIfBetter(const IColumn & column, size_t row_num) { return this->changeIfGreater(column, row_num); }
-    bool changeIfBetter(const Self & to)                        { return this->changeIfGreater(to); }
+    bool changeIfBetter(const IColumn & column, size_t row_num, Arena * arena) { return this->changeIfGreater(column, row_num, arena); }
+    bool changeIfBetter(const Self & to, Arena * arena)                        { return this->changeIfGreater(to, arena); }
 
     static const char * name() { return "max"; }
 };
@@ -595,8 +588,8 @@ struct AggregateFunctionAnyData : Data
 {
     using Self = AggregateFunctionAnyData<Data>;
 
-    bool changeIfBetter(const IColumn & column, size_t row_num) { return this->changeFirstTime(column, row_num); }
-    bool changeIfBetter(const Self & to)                        { return this->changeFirstTime(to); }
+    bool changeIfBetter(const IColumn & column, size_t row_num, Arena * arena) { return this->changeFirstTime(column, row_num, arena); }
+    bool changeIfBetter(const Self & to, Arena * arena)                        { return this->changeFirstTime(to, arena); }
 
     static const char * name() { return "any"; }
 };
@@ -606,8 +599,8 @@ struct AggregateFunctionAnyLastData : Data
 {
     using Self = AggregateFunctionAnyLastData<Data>;
 
-    bool changeIfBetter(const IColumn & column, size_t row_num) { return this->changeEveryTime(column, row_num); }
-    bool changeIfBetter(const Self & to)                        { return this->changeEveryTime(to); }
+    bool changeIfBetter(const IColumn & column, size_t row_num, Arena * arena) { return this->changeEveryTime(column, row_num, arena); }
+    bool changeIfBetter(const Self & to, Arena * arena)                        { return this->changeEveryTime(to, arena); }
 
     static const char * name() { return "anyLast"; }
 };
@@ -625,7 +618,7 @@ struct AggregateFunctionAnyHeavyData : Data
 
     using Self = AggregateFunctionAnyHeavyData<Data>;
 
-    bool changeIfBetter(const IColumn & column, size_t row_num)
+    bool changeIfBetter(const IColumn & column, size_t row_num, Arena * arena)
     {
         if (this->isEqualTo(column, row_num))
         {
@@ -635,7 +628,7 @@ struct AggregateFunctionAnyHeavyData : Data
         {
             if (counter == 0)
             {
-                this->change(column, row_num);
+                this->change(column, row_num, arena);
                 ++counter;
                 return true;
             }
@@ -645,7 +638,7 @@ struct AggregateFunctionAnyHeavyData : Data
         return false;
     }
 
-    bool changeIfBetter(const Self & to)
+    bool changeIfBetter(const Self & to, Arena * arena)
     {
         if (this->isEqualTo(to))
         {
@@ -655,7 +648,7 @@ struct AggregateFunctionAnyHeavyData : Data
         {
             if (counter < to.counter)
             {
-                this->change(to);
+                this->change(to, arena);
                 return true;
             }
             else
@@ -670,9 +663,9 @@ struct AggregateFunctionAnyHeavyData : Data
         writeBinary(counter, buf);
     }
 
-    void read(ReadBuffer & buf, const IDataType & data_type)
+    void read(ReadBuffer & buf, const IDataType & data_type, Arena * arena)
     {
-        Data::read(buf, data_type);
+        Data::read(buf, data_type, arena);
         readBinary(counter, buf);
     }
 
@@ -705,14 +698,14 @@ public:
         return type;
     }
 
-    void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
+    void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena * arena) const override
     {
-        this->data(place).changeIfBetter(*columns[0], row_num);
+        this->data(place).changeIfBetter(*columns[0], row_num, arena);
     }
 
-    void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
+    void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena * arena) const override
     {
-        this->data(place).changeIfBetter(this->data(rhs));
+        this->data(place).changeIfBetter(this->data(rhs), arena);
     }
 
     void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
@@ -720,9 +713,9 @@ public:
         this->data(place).write(buf, *type.get());
     }
 
-    void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
+    void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena * arena) const override
     {
-        this->data(place).read(buf, *type.get());
+        this->data(place).read(buf, *type.get(), arena);
     }
 
     void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override