(ctx context.Context, id int)
| 237 | } |
| 238 | |
| 239 | func (e *ParallelNestedLoopApplyExec) innerWorker(ctx context.Context, id int) { |
| 240 | defer trace.StartRegion(ctx, "ParallelApplyInnerWorker").End() |
| 241 | defer e.handleWorkerPanic(ctx, &e.workerWg) |
| 242 | for { |
| 243 | var chk *chunk.Chunk |
| 244 | select { |
| 245 | case chk = <-e.freeChkCh: |
| 246 | case <-e.exit: |
| 247 | return |
| 248 | } |
| 249 | failpoint.Inject("parallelApplyInnerWorkerPanic", nil) |
| 250 | err := e.fillInnerChunk(ctx, id, chk) |
| 251 | if err == nil && chk.NumRows() == 0 { // no more data, this goroutine can exit |
| 252 | return |
| 253 | } |
| 254 | if e.putResult(chk, err) { |
| 255 | return |
| 256 | } |
| 257 | } |
| 258 | } |
| 259 | |
| 260 | func (e *ParallelNestedLoopApplyExec) putResult(chk *chunk.Chunk, err error) (exit bool) { |
| 261 | select { |
no test coverage detected