Fix forever hanging when HashAgg is called by apply (#12760) (#12766)

executor/aggregate.go

@@ -163,6 +163,7 @@ type HashAggExec struct {
 	isChildReturnEmpty bool
 
 	childResult *chunk.Chunk
+	executed    bool
 }
 
 // HashAggInput indicates the input of hash agg exec.
@@ -216,6 +217,7 @@ func (e *HashAggExec) Close() error {
 	}
 	for range e.finalOutputCh {
 	}
+	e.executed = false
 	return e.baseExecutor.Close()
 }
 
@@ -614,10 +616,14 @@ func (e *HashAggExec) parallelExec(ctx context.Context, chk *chunk.Chunk) error
 		}
 	})
 
+	if e.executed {
+		return nil
+	}
 	for !chk.IsFull() {
 		e.finalInputCh <- chk
 		result, ok := <-e.finalOutputCh
 		if !ok { // all finalWorkers exited
+			e.executed = true
 			if chk.NumRows() > 0 { // but there are some data left
 				return nil
 			}