code: func (e *HashAggExec) execute(ctx context.Context) (err error) {
  	inputIter := chunk.NewIterator4Chunk(e.childResult)
  	for {
  		err := e.children[0].Next(ctx, chunk.NewRecordBatch(e.childResult))
  		if err != nil {
  			return err
  		}
  
  		failpoint.Inject("unparallelHashAggError", func(val failpoint.Value) {
  			if val.(bool) {
  				failpoint.Return(errors.New("HashAggExec.unparallelExec error"))
  			}
  		})
  
  		// no more data.
  		if e.childResult.NumRows() == 0 {
  			return nil
  		}
  		for row := inputIter.Begin(); row != inputIter.End(); row = inputIter.Next() {
  			groupKey, err := e.getGroupKey(row)
  			if err != nil {
  				return err
  			}
  			if !e.groupSet.Exist(groupKey) {
  				e.groupSet.Insert(groupKey)
  				e.groupKeys = append(e.groupKeys, groupKey)
  			}
  			partialResults := e.getPartialResults(groupKey)
  			for i, af := range e.PartialAggFuncs {
  				err = af.UpdatePartialResult(e.ctx, []chunk.Row{row}, partialResults[i])
  				if err != nil {
  					return err
  				}
  			}
  		}
  	}
  }
docs: // execute fetches Chunks from src and update each aggregate function for each row in Chunk.