From 7c41d091d15f764ad3c409c97852007c21c2e7e6 Mon Sep 17 00:00:00 2001
From: Han Fei <hanfei19910905@gmail.com>
Date: Mon, 11 Mar 2019 18:20:18 +0800
Subject: [PATCH] fix decimal bug

---
 .../AggregateFunctions/AggregateFunctionAvg.h | 29 +++++++++++++++++--
 .../AggregateFunctions/AggregateFunctionSum.h | 29 +++++++++++++++++--
 .../AggregateFunctions/IAggregateFunction.h   |  2 ++
 3 files changed, 55 insertions(+), 5 deletions(-)

diff --git a/dbms/src/AggregateFunctions/AggregateFunctionAvg.h b/dbms/src/AggregateFunctions/AggregateFunctionAvg.h
index 18b734885cd..471691328f0 100644
--- a/dbms/src/AggregateFunctions/AggregateFunctionAvg.h
+++ b/dbms/src/AggregateFunctions/AggregateFunctionAvg.h
@@ -44,6 +44,16 @@ class AggregateFunctionAvg final : public IAggregateFunctionDataHelper<Aggregate
     }
     String getName() const override { return "avg"; }
 
+    AggregateDataPtr adjust_place(AggregateDataPtr place) const {
+        size_t pad = (32 - uint64_t(place) % 32) % 32;
+        return place + pad;
+    }
+
+    AggregateDataPtr adjust_place(ConstAggregateDataPtr place) const {
+        size_t pad = (32 - uint64_t(place) % 32) % 32;
+        return (char*)(place + pad);
+    }
+
     DataTypePtr getReturnType() const override
     {
         if constexpr (IsDecimal<T>)
@@ -54,33 +64,48 @@ class AggregateFunctionAvg final : public IAggregateFunctionDataHelper<Aggregate
 
     void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
     {
+        if constexpr (IsDecimal<T>) {
+            place = adjust_place(place);
+        }
         this->data(place).sum += static_cast<const ColumnVector<T> &>(*columns[0]).getData()[row_num];
         ++this->data(place).count;
     }
 
     void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
     {
+        if constexpr (IsDecimal<T>) {
+            place = adjust_place(place);
+            rhs = adjust_place(rhs);
+        }
         this->data(place).sum += this->data(rhs).sum;
         this->data(place).count += this->data(rhs).count;
     }
 
     void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
     {
+        if constexpr (IsDecimal<T>) {
+            place = adjust_place(place);
+        }
         writeBinary(this->data(place).sum, buf);
         writeVarUInt(this->data(place).count, buf);
     }
 
     void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
     {
+        if constexpr (IsDecimal<T>) {
+            place = adjust_place(place);
+        }
         readBinary(this->data(place).sum, buf);
         readVarUInt(this->data(place).count, buf);
     }
 
     void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override
     {
-        if constexpr (IsDecimal<T>)
+        if constexpr (IsDecimal<T>) {
+            place = adjust_place(place);
             static_cast<ColumnDecimal &>(to).getData().push_back(
                 this->data(place).sum.getAvg(this->data(place).count, result_prec, result_scale));
+        }
         else
             static_cast<ColumnFloat64 &>(to).getData().push_back(
                 static_cast<Float64>(this->data(place).sum) / this->data(place).count);
@@ -89,7 +114,7 @@ class AggregateFunctionAvg final : public IAggregateFunctionDataHelper<Aggregate
     void create(AggregateDataPtr place) const override {
         using Data = AggregateFunctionAvgData<typename NearestFieldType<T>::Type>;
         if constexpr (IsDecimal<T>)
-            new (place) Data(result_prec, result_scale);
+            new (adjust_place(place)) Data(result_prec, result_scale);
         else
             new (place) Data;
     }
diff --git a/dbms/src/AggregateFunctions/AggregateFunctionSum.h b/dbms/src/AggregateFunctions/AggregateFunctionSum.h
index 45b66929281..dc45fe10146 100644
--- a/dbms/src/AggregateFunctions/AggregateFunctionSum.h
+++ b/dbms/src/AggregateFunctions/AggregateFunctionSum.h
@@ -107,7 +107,7 @@ class AggregateFunctionSum final : public IAggregateFunctionDataHelper<Data, Agg
     PrecType result_prec;
 
     AggregateFunctionSum(){}
-    
+
     AggregateFunctionSum(PrecType prec, ScaleType scale) {
         SumDecimalInferer::infer(prec, scale, result_prec, result_scale);
     };
@@ -120,35 +120,58 @@ class AggregateFunctionSum final : public IAggregateFunctionDataHelper<Data, Agg
         }
     }
 
+    AggregateDataPtr adjust_place(AggregateDataPtr place) const {
+        size_t pad = (32 - uint64_t(place) % 32) % 32;
+        return place + pad;
+    }
+
+    AggregateDataPtr adjust_place(ConstAggregateDataPtr place) const {
+        size_t pad = (32 - uint64_t(place) % 32) % 32;
+        return (char*)(place + pad);
+    }
+
     void create(AggregateDataPtr place) const override {
-        if constexpr (IsDecimal<T> && IsDecimal<TResult>)
-            new (place) Data(result_prec, result_scale);
+        if constexpr (IsDecimal<T> && IsDecimal<TResult>) {
+            new (adjust_place(place)) Data(result_prec, result_scale);
+        }
         else
             new (place) Data;
     }
 
     void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
     {
+        if constexpr (IsDecimal<T> && IsDecimal<TResult>)
+            place = adjust_place(place);
         this->data(place).add(static_cast<const ColumnVector<T> &>(*columns[0]).getData()[row_num]);
     }
 
     void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
     {
+        if constexpr (IsDecimal<T> && IsDecimal<TResult>) {
+            place = adjust_place(place);
+            rhs = adjust_place(rhs);
+        }
         this->data(place).merge(this->data(rhs));
     }
 
     void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
     {
+        if constexpr (IsDecimal<T> && IsDecimal<TResult>)
+            place = adjust_place(place);
         this->data(place).write(buf);
     }
 
     void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
     {
+        if constexpr (IsDecimal<T> && IsDecimal<TResult>)
+            place = adjust_place(place);
         this->data(place).read(buf);
     }
 
     void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override
     {
+        if constexpr (IsDecimal<T> && IsDecimal<TResult>)
+            place = adjust_place(place);
         static_cast<ColumnVector<TResult> &>(to).getData().push_back(this->data(place).get());
     }
 
diff --git a/dbms/src/AggregateFunctions/IAggregateFunction.h b/dbms/src/AggregateFunctions/IAggregateFunction.h
index 96d03f1be63..b0a72eb81ab 100644
--- a/dbms/src/AggregateFunctions/IAggregateFunction.h
+++ b/dbms/src/AggregateFunctions/IAggregateFunction.h
@@ -154,6 +154,8 @@ class IAggregateFunctionDataHelper : public IAggregateFunctionHelper<Derived>
 
     size_t sizeOfData() const override
     {
+        if constexpr (IsDecimal<T>)
+            return sizeof(Data) + 32;
         return sizeof(Data);
     }