executor: fix data race in parallel sort test (#52246)

close #52238
pingcap · Apr 1, 2024 · 6f0adb8 · 6f0adb8
1 parent 42673cb
commit 6f0adb8
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 4 deletions.
diff --git a/pkg/executor/sortexec/parallel_sort_spill_helper.go b/pkg/executor/sortexec/parallel_sort_spill_helper.go
@@ -177,6 +177,11 @@ func (p *parallelSortSpillHelper) spillImpl(merger *multiWayMerger) error {
 	inDisk.GetDiskTracker().AttachTo(p.sortExec.diskTracker)
 
 	spilledRowChannel := make(chan chunk.Row, 10000)
+
+	// We must wait the finish of the following goroutine,
+	// or we will exit `spillImpl` function in advance and
+	// this will cause data race.
+	defer channel.Clear(spilledRowChannel)
 	go func() {
 		defer func() {
 			if r := recover(); r != nil {
@@ -207,9 +212,6 @@ func (p *parallelSortSpillHelper) spillImpl(merger *multiWayMerger) error {
 	for {
 		select {
 		case <-p.finishCh:
-			// We must wait the finish of the above goroutine,
-			// or p.errOutputChan may be closed in advandce.
-			channel.Clear(spilledRowChannel)
 			return nil
 		case row, ok = <-spilledRowChannel:
 			if !ok {

diff --git a/pkg/executor/sortexec/sort.go b/pkg/executor/sortexec/sort.go
@@ -437,7 +437,6 @@ func (e *SortExec) generateResult(waitGroups ...*util.WaitGroupWrapper) {
 			processPanicAndLog(e.Parallel.resultChannel, r)
 		}
 
-		close(e.Parallel.resultChannel)
 		for i := range e.Parallel.sortedRowsIters {
 			e.Parallel.sortedRowsIters[i].Reset(nil)
 		}
@@ -447,6 +446,7 @@ func (e *SortExec) generateResult(waitGroups ...*util.WaitGroupWrapper) {
 			}
 		}
 		e.Parallel.merger = nil
+		close(e.Parallel.resultChannel)
 	}()
 
 	if !e.Parallel.spillHelper.isSpillTriggered() {