code: func (e *HashJoinExec) fetchOuterChunks(ctx context.Context) {
  	hasWaitedForInner := false
  	for {
  		if e.finished.Load().(bool) {
  			return
  		}
  		var outerResource *outerChkResource
  		var ok bool
  		select {
  		case <-e.closeCh:
  			return
  		case outerResource, ok = <-e.outerChkResourceCh:
  			if !ok {
  				return
  			}
  		}
  		outerResult := outerResource.chk
  		if e.isOuterJoin {
  			required := int(atomic.LoadInt64(&e.requiredRows))
  			outerResult.SetRequiredRows(required, e.maxChunkSize)
  		}
  		err := e.outerExec.Next(ctx, chunk.NewRecordBatch(outerResult))
  		if err != nil {
  			e.joinResultCh <- &hashjoinWorkerResult{
  				err: err,
  			}
  			return
  		}
  		if !hasWaitedForInner {
  			if outerResult.NumRows() == 0 {
  				e.finished.Store(true)
  				return
  			}
  			jobFinished, innerErr := e.wait4Inner()
  			if innerErr != nil {
  				e.joinResultCh <- &hashjoinWorkerResult{
  					err: innerErr,
  				}
  				return
  			} else if jobFinished {
  				return
  			}
  			hasWaitedForInner = true
  		}
  
  		if outerResult.NumRows() == 0 {
  			return
  		}
  		outerResource.dest <- outerResult
  	}
  }
docs: // fetchOuterChunks get chunks from fetches chunks from the big table in a background goroutine
  // and sends the chunks to multiple channels which will be read by multiple join workers.