[BugFix] Fix spill group by limit hang (StarRocks#42583)

Signed-off-by: stdpain <[email protected]>
stdpain · Apr 19, 2024 · c6620c7 · c6620c7
1 parent 3e1cdd5
commit c6620c7
Showing 2 changed files with 3 additions and 0 deletions.
diff --git a/be/src/exec/pipeline/aggregate/spillable_aggregate_blocking_sink_operator.cpp b/be/src/exec/pipeline/aggregate/spillable_aggregate_blocking_sink_operator.cpp
@@ -127,6 +127,7 @@ Status SpillableAggregateBlockingSinkOperator::reset_state(RuntimeState* state,
 }
 
 Status SpillableAggregateBlockingSinkOperator::_spill_all_inputs(RuntimeState* state, const ChunkPtr& chunk) {
+    RETURN_IF(_aggregator->hash_map_variant().size() == 0, Status::OK());
     // spill all data
     DCHECK(!_aggregator->is_none_group_by_exprs());
     _aggregator->hash_map_variant().visit(

diff --git a/be/src/exec/pipeline/aggregate/spillable_aggregate_distinct_blocking_operator.cpp b/be/src/exec/pipeline/aggregate/spillable_aggregate_distinct_blocking_operator.cpp
@@ -94,6 +94,7 @@ Status SpillableAggregateDistinctBlockingSinkOperator::push_chunk(RuntimeState*
     }
     RETURN_IF_ERROR(AggregateDistinctBlockingSinkOperator::push_chunk(state, chunk));
     set_revocable_mem_bytes(_aggregator->hash_set_memory_usage());
+    if (_is_finished) return Status::OK();
     if (_spill_strategy == spill::SpillStrategy::SPILL_ALL) {
         return _spill_all_inputs(state, chunk);
     }
@@ -110,6 +111,7 @@ Status SpillableAggregateDistinctBlockingSinkOperator::reset_state(RuntimeState*
 }
 
 Status SpillableAggregateDistinctBlockingSinkOperator::_spill_all_inputs(RuntimeState* state, const ChunkPtr& chunk) {
+    RETURN_IF(_aggregator->hash_set_variant().size() == 0, Status::OK());
     _aggregator->hash_set_variant().visit(
             [&](auto& hash_set_with_key) { _aggregator->it_hash() = hash_set_with_key->hash_set.begin(); });
     CHECK(!_aggregator->spill_channel()->has_task());