Fix resource cleanup issue after recover (#12542)

## What type of PR is this?

- [ ] API-change
- [X] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:
#11966,
#12385

issue #

## What this PR does / why we need it:

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

pkg/sql/colexec/rightsemi/join.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/common/bitmap"
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
@@ -145,12 +146,18 @@ func (ctr *container) sendLast(ap *Argument, proc *process.Process, analyze proc
 			return true, nil
 		} else {
 			cnt := 1
-			for v := range ap.Channel {
-				ctr.matched.Or(v)
-				cnt++
-				if cnt == int(ap.NumCPU) {
-					close(ap.Channel)
-					break
+			// The original code didn't handle the context correctly and would cause the system to HUNG!
+			for completed := true; completed; {
+				select {
+				case <-proc.Ctx.Done():
+					return true, moerr.NewInternalError(proc.Ctx, "query has been closed early")
+				case v := <-ap.Channel:
+					ctr.matched.Or(v)
+					cnt++
+					if cnt == int(ap.NumCPU) {
+						close(ap.Channel)
+						completed = false
+					}
 				}
 			}
 		}

pkg/sql/compile/compile.go

@@ -72,6 +72,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
 	"github.com/panjf2000/ants/v2"
+	"go.uber.org/zap"
 )
 
 // Note: Now the cost going from stat is actually the number of rows, so we can only estimate a number for the size of each row.
@@ -478,12 +479,13 @@ func (c *Compile) runOnce() error {
 			defer func() {
 				if e := recover(); e != nil {
 					err := moerr.ConvertPanicError(c.ctx, e)
+					getLogger().Error("panic in run",
+						zap.String("error", err.Error()))
 					errC <- err
-					wg.Done()
 				}
+				wg.Done()
 			}()
 			errC <- c.run(scope)
-			wg.Done()
 		})
 	}
 	wg.Wait()

pkg/sql/compile/scope.go

@@ -67,6 +67,8 @@ func (s *Scope) Run(c *Compile) (err error) {
 	defer func() {
 		if e := recover(); e != nil {
 			err = moerr.ConvertPanicError(s.Proc.Ctx, e)
+			getLogger().Error("panic in scope run",
+				zap.String("error", err.Error()))
 		}
 		p.Cleanup(s.Proc, err != nil, err)
 	}()
@@ -113,6 +115,14 @@ func (s *Scope) MergeRun(c *Compile) error {
 		scope := s.PreScopes[i]
 		wg.Add(1)
 		ants.Submit(func() {
+			defer func() {
+				if e := recover(); e != nil {
+					err := moerr.ConvertPanicError(c.ctx, e)
+					getLogger().Error("panic in merge run run",
+						zap.String("error", err.Error()))
+				}
+				wg.Done()
+			}()
 			switch scope.Magic {
 			case Normal:
 				errChan <- scope.Run(c)
@@ -125,7 +135,6 @@ func (s *Scope) MergeRun(c *Compile) error {
 			case Pushdown:
 				errChan <- scope.PushdownRun()
 			}
-			wg.Done()
 		})
 	}
 	defer wg.Wait()

pkg/vm/vm.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
@@ -53,6 +54,7 @@ func Run(ins Instructions, proc *process.Process) (end bool, err error) {
 	defer func() {
 		if e := recover(); e != nil {
 			err = moerr.ConvertPanicError(proc.Ctx, e)
+			logutil.Errorf("panic in vm.Run: %v", err)
 		}
 	}()