Destroy the previous Accumulator for all aggregates

Author: jiangtian

velox/exec/SimpleAggregateAdapter.h

@@ -366,6 +366,14 @@ class SimpleAggregateAdapter : public Aggregate {
       folly::Range<const vector_size_t*> indices) override {
     setAllNulls(groups, indices);
     for (auto i : indices) {
+      if (isInitialized(groups[i])) {
+        if constexpr (accumulator_custom_destroy_) {   
+          auto accumulator = value<typename FUNC::AccumulatorType>(groups[i]);
+          accumulator->destroy(allocator_);
+        } else {
+          destroyAccumulator<typename FUNC::AccumulatorType>(groups[i]);
+        }
+      }
       new (groups[i] + offset_)
           typename FUNC::AccumulatorType(allocator_, fn_.get());
     }

velox/exec/tests/utils/SumNonPODAggregate.cpp

@@ -123,6 +123,9 @@ class SumNonPODAggregate : public Aggregate {
       char** groups,
       folly::Range<const velox::vector_size_t*> indices) override {
     for (auto i : indices) {
+      if (isInitialized(groups[i])) {
+        value<NonPODInt64>(groups[i])->~NonPODInt64();
+      }
       char* group = value<char>(groups[i]);
       VELOX_CHECK_EQ(reinterpret_cast<uintptr_t>(group) % alignment_, 0);
       new (group) NonPODInt64(0);

velox/functions/lib/aggregates/MinMaxAggregateBase.cpp

@@ -467,6 +467,9 @@ class MinMaxAggregateBase : public exec::Aggregate {
       folly::Range<const vector_size_t*> indices) override {
     exec::Aggregate::setAllNulls(groups, indices);
     for (auto i : indices) {
+      if (isInitialized(groups[i])) {
+        value<SingleValueAccumulator>(groups[i])->destroy(allocator_);
+      }
       new (groups[i] + offset_) SingleValueAccumulator();
     }
   }

velox/functions/lib/aggregates/MinMaxByAggregatesBase.h

@@ -497,6 +497,9 @@ class MinMaxByAggregateBase : public exec::Aggregate {
       if constexpr (std::is_same_v<
                         ValueAccumulatorType,
                         SingleValueAccumulator>) {
+        if (isInitialized(group)) {
+          value(group)->destroy(allocator_);
+        }
         new (group + offset_) SingleValueAccumulator();
       } else {
         *value(group) = ValueAccumulatorType();
@@ -507,6 +510,9 @@ class MinMaxByAggregateBase : public exec::Aggregate {
                         SingleValueAccumulator>) {
         *comparisonValue(group) = ComparisonAccumulatorType();
       } else {
+        if (isInitialized(group)) {
+          comparisonValue(group)->destroy(allocator_);
+        }
         new (group + offset_ + sizeof(ValueAccumulatorType))
             SingleValueAccumulator();
       }

velox/functions/lib/aggregates/SetBaseAggregate.h

@@ -201,6 +201,9 @@ class SetBaseAggregate : public exec::Aggregate {
     const auto& type = resultType()->childAt(0);
     exec::Aggregate::setAllNulls(groups, indices);
     for (auto i : indices) {
+      if (isInitialized(groups[i])) {
+        value(groups[i])->free(*allocator_);
+      }
       new (groups[i] + offset_) AccumulatorType(type, allocator_);
     }
   }

velox/functions/prestosql/aggregates/ApproxPercentileAggregate.cpp

@@ -489,6 +489,9 @@ class ApproxPercentileAggregate : public exec::Aggregate {
     exec::Aggregate::setAllNulls(groups, indices);
     for (auto i : indices) {
       auto group = groups[i];
+      if (isInitialized(group)) {
+        value<KllSketchAccumulator<T>>(group)->~KllSketchAccumulator<T>();
+      }
       new (group + offset_)
           KllSketchAccumulator<T>(allocator_, fixedRandomSeed_);
     }

velox/functions/prestosql/aggregates/ArbitraryAggregate.cpp

@@ -375,6 +375,15 @@ class NonNumericArbitrary : public exec::Aggregate {
       char** groups,
       folly::Range<const vector_size_t*> indices) override {
     for (auto i : indices) {
+      if (isInitialized(groups[i])) {
+        if (clusteredInput_) {
+          auto* accumulator = value<ClusteredNonNumericAccumulator>(groups[i]);
+          std::destroy_at(accumulator);
+        } else {
+          auto* accumulator = value<SingleValueAccumulator>(groups[i]);
+          accumulator->destroy(allocator_);
+        }
+      }
       if (clusteredInput_) {
         new (groups[i] + offset_) ClusteredNonNumericAccumulator();
       } else {

velox/functions/prestosql/aggregates/ArrayAggAggregate.cpp

@@ -331,6 +331,14 @@ class ArrayAggAggregate : public exec::Aggregate {
       char** groups,
       folly::Range<const vector_size_t*> indices) override {
     for (auto index : indices) {
+      if (isInitialized(groups[index])) {
+        if (clusteredInput_) {
+          auto* accumulator = value<ClusteredAccumulator>(groups[index]);
+          std::destroy_at(accumulator);
+        } else {
+          value<ArrayAccumulator>(groups[index])->elements.free(allocator_);
+        }
+      }
       if (clusteredInput_) {
         new (groups[index] + offset_) ClusteredAccumulator();
       } else {

velox/functions/prestosql/aggregates/ClassificationAggregation.cpp

@@ -582,6 +582,9 @@ class ClassificationAggregation : public exec::Aggregate {
     exec::Aggregate::setAllNulls(groups, indices);
     for (auto i : indices) {
       auto group = groups[i];
+      if (isInitialized(group)) {
+        destroyAccumulator<Accumulator<type>>(group);
+      }
       new (group + offset_) Accumulator<type>(allocator_);
     }
   }

velox/functions/prestosql/aggregates/HistogramAggregate.cpp

@@ -528,6 +528,9 @@ class HistogramAggregate : public exec::Aggregate {
       folly::Range<const vector_size_t*> indices) override {
     const auto& type = resultType()->childAt(0);
     for (auto index : indices) {
+      if (isInitialized(groups[index])) {
+        value<AccumulatorType>(groups[index])->free(*allocator_);
+      }
       new (groups[index] + offset_) AccumulatorType{type, allocator_};
     }
   }