code: func (e *HashAggExec) parallelExec(ctx context.Context, chk *chunk.Chunk) error {
  	if !e.prepared {
  		e.prepare4ParallelExec(ctx)
  		e.prepared = true
  	}
  
  	failpoint.Inject("parallelHashAggError", func(val failpoint.Value) {
  		if val.(bool) {
  			failpoint.Return(errors.New("HashAggExec.parallelExec error"))
  		}
  	})
  
  	for !chk.IsFull() {
  		e.finalInputCh <- chk
  		result, ok := <-e.finalOutputCh
  		if !ok { // all finalWorkers exited
  			if chk.NumRows() > 0 { // but there are some data left
  				return nil
  			}
  			if e.isChildReturnEmpty && e.defaultVal != nil {
  				chk.Append(e.defaultVal, 0, 1)
  			}
  			e.isChildReturnEmpty = false
  			return nil
  		}
  		if result.err != nil {
  			return result.err
  		}
  		if chk.NumRows() > 0 {
  			e.isChildReturnEmpty = false
  		}
  	}
  	return nil
  }
docs: // HashAggExec employs one input reader, M partial workers and N final workers to execute parallelly.
  // The parallel execution flow is:
  // 1. input reader reads data from child executor and send them to partial workers.
  // 2. partial worker receives the input data, updates the partial results, and shuffle the partial results to the final workers.
  // 3. final worker receives partial results from all the partial workers, evaluates the final results and sends the final results to the main thread.