refactor: Simplify the partial full handling for distinct aggregation

Summary:
The current partial full is cleared when there is no new distinct in add input if it has been set.
The reason for doing this is to avoid distinct partial aggregation hanging problem
as we only reset the grouping set in case of partial full if there is new distinct found in get output.
This is a bit tricky and it is better to handle reset partial full unconditionally in get output for
distinct type of aggregation.

Differential Revision: D67288365

velox/exec/HashAggregation.cpp

@@ -205,11 +205,6 @@ void HashAggregation::addInput(RowVectorPtr input) {
     if (newDistincts_) {
       // Save input to use for output in getOutput().
       input_ = input;
-    } else {
-      // If no new distinct groups (meaning we don't have anything to output),
-      // then we need to ensure we 'need input'. For that we need to reset
-      // the 'partial full' flag.
-      partialFull_ = false;
     }
   }
 }
@@ -348,7 +343,11 @@ RowVectorPtr HashAggregation::getOutput() {
   }
 
   if (isDistinct_) {
-    return getDistinctOutput();
+    auto output = getDistinctOutput();
+    if (partialFull_) {
+      resetPartialOutputIfNeed();
+    }
+    return output;
   }
 
   const auto& queryConfig = operatorCtx_->driverCtx()->queryConfig();
@@ -393,8 +392,6 @@ RowVectorPtr HashAggregation::getDistinctOutput() {
     // Drop reference to input_ to make it singly-referenced at the producer and
     // allow for memory reuse.
     input_ = nullptr;
-
-    resetPartialOutputIfNeed();
     return output;
   }
   VELOX_CHECK(!newDistincts_);