executor: fix hash join sqllogic test block, data race. (pingcap#8392)

executor/join.go

@@ -262,6 +262,8 @@ func (e *HashJoinExec) fetchInnerRows(ctx context.Context, chkCh chan<- *chunk.C
 		select {
 		case <-doneCh:
 			return
+		case <-e.closeCh:
+			return
 		default:
 			if e.finished.Load().(bool) {
 				return
@@ -275,7 +277,12 @@ func (e *HashJoinExec) fetchInnerRows(ctx context.Context, chkCh chan<- *chunk.C
 			if chk.NumRows() == 0 {
 				return
 			}
-			chkCh <- chk
+			select {
+			case chkCh <- chk:
+				break
+			case <-e.closeCh:
+				return
+			}
 			e.innerResult.Add(chk)
 		}
 	}
@@ -527,19 +534,14 @@ func (e *HashJoinExec) fetchInnerAndBuildHashTable(ctx context.Context) {
 	doneCh := make(chan struct{})
 	go util.WithRecovery(func() { e.fetchInnerRows(ctx, innerResultCh, doneCh) }, nil)
 
-	if e.finished.Load().(bool) {
-		return
-	}
 	// TODO: Parallel build hash table. Currently not support because `mvmap` is not thread-safe.
 	err := e.buildHashTableForList(innerResultCh)
 	if err != nil {
 		e.innerFinished <- errors.Trace(err)
 		close(doneCh)
-		// fetchInnerRows may be blocked by this channel, so read from the channel to unblock it.
-		select {
-		case <-innerResultCh:
-		default:
-		}
+	}
+	// wait fetchInnerRows be finished.
+	for range innerResultCh {
 	}
 }