code: func (e *HashJoinExec) Next(ctx context.Context, req *chunk.RecordBatch) (err error) {
  	if e.runtimeStats != nil {
  		start := time.Now()
  		defer func() { e.runtimeStats.Record(time.Since(start), req.NumRows()) }()
  	}
  	if !e.prepared {
  		e.innerFinished = make(chan error, 1)
  		go util.WithRecovery(func() { e.fetchInnerAndBuildHashTable(ctx) }, e.handleFetchInnerAndBuildHashTablePanic)
  		e.fetchOuterAndProbeHashTable(ctx)
  		e.prepared = true
  	}
  	if e.isOuterJoin {
  		atomic.StoreInt64(&e.requiredRows, int64(req.RequiredRows()))
  	}
  	req.Reset()
  	if e.joinResultCh == nil {
  		return nil
  	}
  	result, ok := <-e.joinResultCh
  	if !ok {
  		return nil
  	}
  	if result.err != nil {
  		e.finished.Store(true)
  		return result.err
  	}
  	req.SwapColumns(result.chk)
  	result.src <- result.chk
  	return nil
  }
docs: // Next implements the Executor Next interface.
  // hash join constructs the result following these steps:
  // step 1. fetch data from inner child and build a hash table;
  // step 2. fetch data from outer child in a background goroutine and probe the hash table in multiple join workers.