code: func (e *HashJoinExec) Open(ctx context.Context) error {
  	if err := e.baseExecutor.Open(ctx); err != nil {
  		return err
  	}
  
  	e.prepared = false
  	e.memTracker = memory.NewTracker(e.id, e.ctx.GetSessionVars().MemQuotaHashJoin)
  	e.memTracker.AttachTo(e.ctx.GetSessionVars().StmtCtx.MemTracker)
  
  	e.hashTableValBufs = make([][][]byte, e.concurrency)
  	e.hashJoinBuffers = make([]*hashJoinBuffer, 0, e.concurrency)
  	for i := uint(0); i < e.concurrency; i++ {
  		buffer := &hashJoinBuffer{
  			data:  make([]types.Datum, len(e.outerKeys)),
  			bytes: make([]byte, 0, 10000),
  		}
  		e.hashJoinBuffers = append(e.hashJoinBuffers, buffer)
  	}
  
  	e.innerKeyColIdx = make([]int, len(e.innerKeys))
  	for i := range e.innerKeys {
  		e.innerKeyColIdx[i] = e.innerKeys[i].Index
  	}
  
  	e.closeCh = make(chan struct{})
  	e.finished.Store(false)
  	e.joinWorkerWaitGroup = sync.WaitGroup{}
  	return nil
  }
docs: // Open implements the Executor Open interface.