fix: Fix crash from hash aggregation row container cleanup due to abort (facebookincubator#13979)

Summary:
Pull Request resolved: facebookincubator#13979

Aggregates could be uninitialized when destroying, causing process to crash. Avoid destroying uninitialized aggregates.

Reviewed By: skyelves, kewang1024

Differential Revision: D77632466

fbshipit-source-id: 155cef4a7dd2a9cc09b831218a1e5ee31bd2198c

Author: tanjialiang

velox/exec/DistinctAggregations.cpp

@@ -49,6 +49,9 @@ class TypedDistinctAggregations : public DistinctAggregations {
         },
         [this](folly::Range<char**> groups) {
           for (auto* group : groups) {
+            if (!isInitialized(group)) {
+              continue;
+            }
             auto* accumulator =
                 reinterpret_cast<AccumulatorType*>(group + offset_);
             accumulator->free(*allocator_);

velox/exec/DistinctAggregations.h

@@ -101,6 +101,10 @@ class DistinctAggregations {
       char** groups,
       folly::Range<const vector_size_t*> indices) = 0;
 
+  bool isInitialized(char* group) const {
+    return group[initializedByte_] & initializedMask_;
+  }
+
   HashStringAllocator* allocator_;
   int32_t offset_;
   int32_t nullByte_;

velox/exec/tests/AggregationTest.cpp

@@ -4030,6 +4030,64 @@ TEST_F(AggregationTest, destroyAfterPartialInitialization) {
   ASSERT_TRUE(agg.destroyCalled);
 }
 
+DEBUG_ONLY_TEST_F(
+    AggregationTest,
+    uninitializedDistinctAggrWithExternalMemAggrDuringAbort) {
+  const auto createInput =
+      [&](int32_t startKey, uint32_t numGroups, uint32_t numElementsPerGroup) {
+        return makeRowVector({
+            makeFlatVector<int32_t>([&]() {
+              std::vector<int32_t> keys;
+              for (auto i = 0; i < numGroups; ++i) {
+                for (auto j = 0; j < numElementsPerGroup; ++j) {
+                  keys.push_back(startKey + i);
+                }
+              }
+              return keys;
+            }()),
+            makeFlatVector<int32_t>(
+                numGroups * numElementsPerGroup,
+                [&](auto row) { return startKey; }),
+        });
+      };
+
+  std::vector<RowVectorPtr> inputs;
+  inputs.emplace_back(createInput(0, 10000, 10));
+  createDuckDbTable(inputs);
+
+  GroupingSet* groupingSet{nullptr};
+
+  std::atomic_bool groupingSetExtracted{false};
+  SCOPED_TESTVALUE_SET(
+      "facebook::velox::exec::GroupingSet::addInputForActiveRows",
+      std::function<void(GroupingSet*)>([&](GroupingSet* _groupingSet) {
+        if (!groupingSetExtracted.exchange(true)) {
+          groupingSet = _groupingSet;
+        }
+      }));
+
+  SCOPED_TESTVALUE_SET(
+      "facebook::velox::memory::MemoryPoolImpl::reserveThreadSafe",
+      std::function<void(void*)>([&](void* /*unused*/) {
+        if (groupingSet == nullptr) {
+          return;
+        }
+        if (groupingSet->numRows() > 0) {
+          VELOX_FAIL("Inject allocation failure.");
+        }
+      }));
+
+  auto plan = PlanBuilder()
+                  .values(inputs)
+                  .singleAggregation({"c0"}, {"array_agg(distinct c1)"})
+                  .planNode();
+
+  VELOX_ASSERT_THROW(
+      assertQuery(
+          plan, "SELECT c0, array_agg(distinct c1) FROM tmp GROUP BY c0"),
+      "Inject allocation failure.");
+}
+
 TEST_F(AggregationTest, nanKeys) {
   // Some keys are NaNs.
   auto kNaN = std::numeric_limits<double>::quiet_NaN();