Repository: matrixorigin/matrixone
Pull request #5701 (2022-10-18 03:57:56)

Revert "refactor pipeline's clean up method. (#5634)"
This reverts commit 4484abc5344e1e6a0e63321485547402568185f2.

## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:


Patch:
diff --git a/pkg/container/batch/batch.go b/pkg/container/batch/batch.go
index 81dbf5e7f..69e5cc54a 100644
--- a/pkg/container/batch/batch.go
+++ b/pkg/container/batch/batch.go
@@ -256,13 +256,12 @@ func (bat *Batch) Clean(m *mpool.MPool) {
 
 func (bat *Batch) String() string {
 	var buf bytes.Buffer
-	buf.WriteString(fmt.Sprintf("ReadOnly: %v\n", bat.Ro))
-	buf.WriteString(fmt.Sprintf("Cnt: %d\n", bat.Cnt))
-	buf.WriteString(fmt.Sprintf("Attr: %s\n", bat.Attrs))
+
 	for i, vec := range bat.Vecs {
-		buf.WriteString(fmt.Sprintf("vector %d :\n", i+1))
-		buf.WriteString(vec.String())
-		buf.WriteByte('\n')
+		buf.WriteString(fmt.Sprintf("%v\n", i))
+		if len(bat.Zs) > 0 {
+			buf.WriteString(fmt.Sprintf("\t%s\n", vec))
+		}
 	}
 	return buf.String()
 }
diff --git a/pkg/sql/colexec/anti/join.go b/pkg/sql/colexec/anti/join.go
index f69aaf08f..d7e1265e0 100644
--- a/pkg/sql/colexec/anti/join.go
+++ b/pkg/sql/colexec/anti/join.go
@@ -49,6 +49,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 		case Build:
 			if err := ctr.build(ap, proc, anal); err != nil {
 				ctr.state = End
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
 				return true, err
 			}
 			ctr.state = Probe
@@ -56,6 +59,12 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -64,12 +73,18 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			if ctr.bat == nil || ctr.bat.Length() == 0 {
 				if err := ctr.emptyProbe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
 					proc.SetInputBatch(nil)
 					return true, err
 				}
 			} else {
 				if err := ctr.probe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
 					proc.SetInputBatch(nil)
 					return true, err
 				}
@@ -138,6 +153,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 	if err := ctr.evalJoinCondition(bat, ap.Conditions[0], proc); err != nil {
 		return err
 	}
+	defer ctr.freeJoinCondition(proc)
 	count := bat.Length()
 	mSels := ctr.mp.Sels()
 	itr := ctr.mp.Map().NewIterator()
@@ -201,6 +217,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	for i, cond := range conds {
 		vec, err := colexec.EvalExpr(bat, proc, cond)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.Mp())
+				}
+			}
 			return err
 		}
 		ctr.vecs[i] = vec
@@ -215,3 +236,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	}
 	return nil
 }
+
+func (ctr *container) freeJoinCondition(proc *process.Process) {
+	for i := range ctr.evecs {
+		if ctr.evecs[i].needFree {
+			ctr.evecs[i].vec.Free(proc.Mp())
+		}
+	}
+}
diff --git a/pkg/sql/colexec/anti/types.go b/pkg/sql/colexec/anti/types.go
index 977e8f7c8..bfb10897d 100644
--- a/pkg/sql/colexec/anti/types.go
+++ b/pkg/sql/colexec/anti/types.go
@@ -16,12 +16,10 @@ package anti
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -59,36 +57,3 @@ type Argument struct {
 	Cond       *plan.Expr
 	Conditions [][]*plan.Expr
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-		ctr.cleanEvalVectors(mp)
-		ctr.cleanHashMap()
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.mp != nil {
-		ctr.mp.Free()
-		ctr.mp = nil
-	}
-}
-
-func (ctr *container) cleanEvalVectors(mp *mpool.MPool) {
-	for i := range ctr.evecs {
-		if ctr.evecs[i].needFree && ctr.evecs[i].vec != nil {
-			ctr.evecs[i].vec.Free(mp)
-			ctr.evecs[i].vec = nil
-		}
-	}
-}
diff --git a/pkg/sql/colexec/connector/connector.go b/pkg/sql/colexec/connector/connector.go
index db042fd52..b3f484768 100644
--- a/pkg/sql/colexec/connector/connector.go
+++ b/pkg/sql/colexec/connector/connector.go
@@ -34,7 +34,12 @@ func Call(_ int, proc *process.Process, arg any) (bool, error) {
 	reg := ap.Reg
 	bat := proc.InputBatch()
 	if bat == nil {
-		return true, nil
+		select {
+		case <-reg.Ctx.Done():
+			return true, nil
+		case reg.Ch <- bat:
+			return true, nil
+		}
 	}
 	if bat.Length() == 0 {
 		return false, nil
diff --git a/pkg/sql/colexec/connector/types.go b/pkg/sql/colexec/connector/types.go
index 4f7041975..a16fb6812 100644
--- a/pkg/sql/colexec/connector/types.go
+++ b/pkg/sql/colexec/connector/types.go
@@ -22,20 +22,3 @@ import (
 type Argument struct {
 	Reg *process.WaitRegister
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	if pipelineFailed {
-		for len(arg.Reg.Ch) > 0 {
-			bat := <-arg.Reg.Ch
-			if bat == nil {
-				break
-			}
-			bat.Clean(proc.Mp())
-		}
-	}
-
-	select {
-	case arg.Reg.Ch <- nil:
-	case <-arg.Reg.Ctx.Done():
-	}
-}
diff --git a/pkg/sql/colexec/deletion/deletion.go b/pkg/sql/colexec/deletion/deletion.go
index 17b456fc6..3e3de5b12 100644
--- a/pkg/sql/colexec/deletion/deletion.go
+++ b/pkg/sql/colexec/deletion/deletion.go
@@ -74,7 +74,7 @@ func Call(_ int, proc *process.Process, arg any) (bool, error) {
 			err := p.DeleteCtxs[i].TableSource.Delete(ctx, tmpBat, p.DeleteCtxs[i].UseDeleteKey)
 			if err != nil {
 				tmpBat.Clean(proc.Mp())
-				return true, err
+				return false, err
 			}
 			affectedRows += cnt
 
@@ -93,7 +93,7 @@ func Call(_ int, proc *process.Process, arg any) (bool, error) {
 			err := p.DeleteCtxs[i].TableSource.Delete(ctx, tmpBat, p.DeleteCtxs[i].UseDeleteKey)
 			if err != nil {
 				tmpBat.Clean(proc.Mp())
-				return true, err
+				return false, err
 			}
 			affectedRows += uint64(batLen)
 
diff --git a/pkg/sql/colexec/deletion/types.go b/pkg/sql/colexec/deletion/types.go
index 5e6be2668..86427e6bf 100644
--- a/pkg/sql/colexec/deletion/types.go
+++ b/pkg/sql/colexec/deletion/types.go
@@ -17,7 +17,6 @@ package deletion
 import (
 	plan2 "github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 type Argument struct {
@@ -36,6 +35,3 @@ type DeleteCtx struct {
 	ComputeIndexInfos  []*plan2.ComputeIndexInfo
 	IndexAttrs         []string
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/dispatch/dispatch.go b/pkg/sql/colexec/dispatch/dispatch.go
index 10b43155a..dc971495a 100644
--- a/pkg/sql/colexec/dispatch/dispatch.go
+++ b/pkg/sql/colexec/dispatch/dispatch.go
@@ -37,6 +37,12 @@ func Call(_ int, proc *process.Process, arg any) (bool, error) {
 	ap := arg.(*Argument)
 	bat := proc.InputBatch()
 	if bat == nil {
+		for _, reg := range ap.Regs {
+			select {
+			case <-reg.Ctx.Done():
+			case reg.Ch <- nil:
+			}
+		}
 		return true, nil
 	}
 	vecs := ap.vecs[:0]
@@ -44,7 +50,7 @@ func Call(_ int, proc *process.Process, arg any) (bool, error) {
 		if bat.Vecs[i].IsOriginal() {
 			vec, err := vector.Dup(bat.Vecs[i], proc.Mp())
 			if err != nil {
-				return true, err
+				return false, err
 			}
 			vecs = append(vecs, vec)
 		}
@@ -86,5 +92,5 @@ func Call(_ int, proc *process.Process, arg any) (bool, error) {
 			return false, nil
 		}
 	}
-	return false, nil
+	return true, nil
 }
diff --git a/pkg/sql/colexec/dispatch/types.go b/pkg/sql/colexec/dispatch/types.go
index d3c62a8bd..93b864d75 100644
--- a/pkg/sql/colexec/dispatch/types.go
+++ b/pkg/sql/colexec/dispatch/types.go
@@ -29,24 +29,3 @@ type Argument struct {
 	vecs []*vector.Vector
 	Regs []*process.WaitRegister
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	if pipelineFailed {
-		for i := range arg.Regs {
-			for len(arg.Regs[i].Ch) > 0 {
-				bat := <-arg.Regs[i].Ch
-				if bat == nil {
-					break
-				}
-				bat.Clean(proc.Mp())
-			}
-		}
-	}
-
-	for i := range arg.Regs {
-		select {
-		case <-arg.Regs[i].Ctx.Done():
-		case arg.Regs[i].Ch <- nil:
-		}
-	}
-}
diff --git a/pkg/sql/colexec/expr_eval.go b/pkg/sql/colexec/expr_eval.go
index 06c298080..f991b2004 100644
--- a/pkg/sql/colexec/expr_eval.go
+++ b/pkg/sql/colexec/expr_eval.go
@@ -40,8 +40,6 @@ var (
 	constTimestampType  = types.Type{Oid: types.T_timestamp}
 )
 
-// EvalExpr return a vector corresponding to an expression.
-// if vector not one of batch's vectors. generate it.
 func EvalExpr(bat *batch.Batch, proc *process.Process, expr *plan.Expr) (*vector.Vector, error) {
 	var vec *vector.Vector
 
@@ -110,17 +108,17 @@ func EvalExpr(bat *batch.Batch, proc *process.Process, expr *plan.Expr) (*vector
 			return nil, err
 		}
 		vs := make([]*vector.Vector, len(t.F.Args))
-		mp := make(map[*vector.Vector]struct{})
-		for i := range bat.Vecs {
-			mp[bat.Vecs[i]] = struct{}{}
-		}
 		for i := range vs {
 			v, err := EvalExpr(bat, proc, t.F.Args[i])
 			if err != nil {
 				if proc != nil {
+					mp := make(map[*vector.Vector]uint8)
+					for i := range bat.Vecs {
+						mp[bat.Vecs[i]] = 0
+					}
 					for j := 0; j < i; j++ {
 						if _, ok := mp[vs[j]]; !ok {
-							vs[j].Free(proc.Mp())
+							vector.Clean(vs[j], proc.Mp())
 						}
 					}
 				}
@@ -128,13 +126,20 @@ func EvalExpr(bat *batch.Batch, proc *process.Process, expr *plan.Expr) (*vector
 			}
 			vs[i] = v
 		}
-
-		vec, err = f.VecFn(vs, proc)
-		for i := range vs {
-			if _, ok := mp[vs[i]]; !ok {
-				vs[i].Free(proc.Mp())
+		defer func() {
+			if proc != nil {
+				mp := make(map[*vector.Vector]uint8)
+				for i := range bat.Vecs {
+					mp[bat.Vecs[i]] = 0
+				}
+				for i := range vs {
+					if _, ok := mp[vs[i]]; !ok {
+						vector.Clean(vs[i], proc.Mp())
+					}
+				}
 			}
-		}
+		}()
+		vec, err = f.VecFn(vs, proc)
 		if err != nil {
 			return nil, err
 		}
@@ -209,16 +214,13 @@ func JoinFilterEvalExpr(r, s *batch.Batch, rRow int, proc *process.Process, expr
 			return nil, err
 		}
 		vs := make([]*vector.Vector, len(t.F.Args))
-		mp := make(map[*vector.Vector]struct{})
-		for i := range r.Vecs {
-			mp[r.Vecs[i]] = struct{}{}
-		}
-		for i := range s.Vecs {
-			mp[s.Vecs[i]] = struct{}{}
-		}
 		for i := range vs {
 			v, err := JoinFilterEvalExpr(r, s, rRow, proc, t.F.Args[i])
 			if err != nil {
+				mp := make(map[*vector.Vector]uint8)
+				for i := range s.Vecs {
+					mp[s.Vecs[i]] = 0
+				}
 				for j := 0; j < i; j++ {
 					if _, ok := mp[vs[j]]; !ok {
 						vector.Clean(vs[j], proc.Mp())
@@ -228,12 +230,18 @@ func JoinFilterEvalExpr(r, s *batch.Batch, rRow int, proc *process.Process, expr
 			}
 			vs[i] = v
 		}
-		vec, err = f.VecFn(vs, proc)
-		for i := range vs {
-			if _, ok := mp[vs[i]]; !ok {
-				vector.Clean(vs[i], proc.Mp())
+		defer func() {
+			mp := make(map[*vector.Vector]uint8)
+			for i := range s.Vecs {
+				mp[s.Vecs[i]] = 0
 			}
-		}
+			for i := range vs {
+				if _, ok := mp[vs[i]]; !ok {
+					vector.Clean(vs[i], proc.Mp())
+				}
+			}
+		}()
+		vec, err = f.VecFn(vs, proc)
 		if err != nil {
 			return nil, err
 		}
diff --git a/pkg/sql/colexec/external/types.go b/pkg/sql/colexec/external/types.go
index 2e29aa4d9..1c0fc6605 100644
--- a/pkg/sql/colexec/external/types.go
+++ b/pkg/sql/colexec/external/types.go
@@ -17,7 +17,6 @@ package external
 import (
 	"context"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 	"io"
 	"sync/atomic"
 
@@ -62,6 +61,3 @@ type ParseLineHandler struct {
 	//simd csv
 	simdCsvLineArray [][]string
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/generate_series/types.go b/pkg/sql/colexec/generate_series/types.go
index a7a1f35b3..62da8bc4d 100644
--- a/pkg/sql/colexec/generate_series/types.go
+++ b/pkg/sql/colexec/generate_series/types.go
@@ -17,7 +17,6 @@ package generate_series
 import (
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 type Number interface {
@@ -36,10 +35,6 @@ type Argument struct {
 	Es *Param
 }
 
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-
-}
-
 var (
 
 //	timeStampCol = &plan.ColDef{
diff --git a/pkg/sql/colexec/group/group.go b/pkg/sql/colexec/group/group.go
index 133a036fc..12e48b5de 100644
--- a/pkg/sql/colexec/group/group.go
+++ b/pkg/sql/colexec/group/group.go
@@ -99,7 +99,7 @@ func (ctr *container) process(ap *Argument, proc *process.Process, anal process.
 	if err := ctr.evalAggVector(bat, ap.Aggs, proc); err != nil {
 		return false, err
 	}
-	defer ctr.cleanAggVectors(proc.Mp())
+	defer ctr.freeAggVector(proc)
 	if ctr.bat == nil {
 		var err error
 
@@ -109,11 +109,13 @@ func (ctr *container) process(ap *Argument, proc *process.Process, anal process.
 		ctr.bat.Aggs = make([]agg.Agg[any], len(ap.Aggs))
 		for i, ag := range ap.Aggs {
 			if ctr.bat.Aggs[i], err = agg.New(ag.Op, ag.Dist, ctr.aggVecs[i].vec.Typ); err != nil {
+				ctr.bat = nil
 				return false, err
 			}
 		}
 		for _, ag := range ctr.bat.Aggs {
 			if err := ag.Grows(1, proc.Mp()); err != nil {
+				ctr.bat.Clean(proc.Mp())
 				return false, err
 			}
 		}
@@ -122,6 +124,7 @@ func (ctr *container) process(ap *Argument, proc *process.Process, anal process.
 		return false, nil
 	}
 	if err := ctr.processH0(bat, ap, proc); err != nil {
+		ctr.bat.Clean(proc.Mp())
 		return false, err
 	}
 	return false, nil
@@ -134,9 +137,11 @@ func (ctr *container) processWithGroup(ap *Argument, proc *process.Process, anal
 	if bat == nil {
 		if ctr.bat != nil {
 			if ap.NeedEval {
-				for i, ag := range ctr.bat.Aggs {
-					vec, err := ag.Eval(proc.Mp())
+				for i, agg := range ctr.bat.Aggs {
+					vec, err := agg.Eval(proc.Mp())
 					if err != nil {
+						ctr.clean()
+						ctr.bat.Clean(proc.Mp())
 						return false, err
 					}
 					ctr.bat.Aggs[i] = nil
@@ -147,6 +152,7 @@ func (ctr *container) processWithGroup(ap *Argument, proc *process.Process, anal
 					ctr.bat.Zs[i] = 1
 				}
 			}
+			ctr.clean()
 			ctr.bat.ExpandNulls()
 			anal.Output(ctr.bat)
 			proc.SetInputBatch(ctr.bat)
@@ -166,9 +172,11 @@ func (ctr *container) processWithGroup(ap *Argument, proc *process.Process, anal
 		ctr.aggVecs = make([]evalVector, len(ap.Aggs))
 	}
 	if err := ctr.evalAggVector(bat, ap.Aggs, proc); err != nil {
+		ctr.clean()
+		ctr.cleanBatch(proc)
 		return false, err
 	}
-	defer ctr.cleanAggVectors(proc.Mp())
+	defer ctr.freeAggVector(proc)
 	if len(ctr.groupVecs) == 0 {
 		ctr.vecs = make([]*vector.Vector, len(ap.Exprs))
 		ctr.groupVecs = make([]evalVector, len(ap.Exprs))
@@ -176,6 +184,13 @@ func (ctr *container) processWithGroup(ap *Argument, proc *process.Process, anal
 	for i, expr := range ap.Exprs {
 		vec, err := colexec.EvalExpr(bat, proc, expr)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.groupVecs[j].needFree {
+					ctr.groupVecs[i].vec.Free(proc.Mp())
+				}
+			}
+			ctr.clean()
+			ctr.cleanBatch(proc)
 			return false, err
 		}
 		ctr.groupVecs[i].vec = vec
@@ -188,6 +203,13 @@ func (ctr *container) processWithGroup(ap *Argument, proc *process.Process, anal
 		}
 		ctr.vecs[i] = vec
 	}
+	defer func() {
+		for i := range ctr.groupVecs {
+			if ctr.groupVecs[i].needFree {
+				ctr.groupVecs[i].vec.Free(proc.Mp())
+			}
+		}
+	}()
 	if ctr.bat == nil {
 		size := 0
 		ctr.bat = batch.NewWithSize(len(ap.Exprs))
@@ -221,11 +243,13 @@ func (ctr *container) processWithGroup(ap *Argument, proc *process.Process, anal
 		case size <= 8:
 			ctr.typ = H8
 			if ctr.intHashMap, err = hashmap.NewIntHashMap(true, ap.Ibucket, ap.Nbucket, proc.Mp()); err != nil {
+				ctr.cleanBatch(proc)
 				return false, err
 			}
 		default:
 			ctr.typ = HStr
 			if ctr.strHashMap, err = hashmap.NewStrMap(true, ap.Ibucket, ap.Nbucket, proc.Mp()); err != nil {
+				ctr.cleanBatch(proc)
 				return false, err
 			}
 		}
@@ -237,6 +261,8 @@ func (ctr *container) processWithGroup(ap *Argument, proc *process.Process, anal
 		err = ctr.processHStr(bat, proc)
 	}
 	if err != nil {
+		ctr.clean()
+		ctr.cleanBatch(proc)
 		return false, err
 	}
 	return false, err
@@ -246,11 +272,8 @@ func (ctr *container) processH0(bat *batch.Batch, ap *Argument, proc *process.Pr
 	for _, z := range bat.Zs {
 		ctr.bat.Zs[0] += z
 	}
-	for i, ag := range ctr.bat.Aggs {
-		err := ag.BulkFill(0, bat.Zs, []*vector.Vector{ctr.aggVecs[i].vec})
-		if err != nil {
-			return err
-		}
+	for i, agg := range ctr.bat.Aggs {
+		agg.BulkFill(0, bat.Zs, []*vector.Vector{ctr.aggVecs[i].vec})
 	}
 	return nil
 }
@@ -341,6 +364,11 @@ func (ctr *container) evalAggVector(bat *batch.Batch, aggs []agg.Aggregate, proc
 	for i, ag := range aggs {
 		vec, err := colexec.EvalExpr(bat, proc, ag.E)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.aggVecs[j].needFree {
+					ctr.aggVecs[i].vec.Free(proc.Mp())
+				}
+			}
 			return err
 		}
 		ctr.aggVecs[i].vec = vec
@@ -354,3 +382,29 @@ func (ctr *container) evalAggVector(bat *batch.Batch, aggs []agg.Aggregate, proc
 	}
 	return nil
 }
+
+func (ctr *container) freeAggVector(proc *process.Process) {
+	for i := range ctr.aggVecs {
+		if ctr.aggVecs[i].needFree {
+			ctr.aggVecs[i].vec.Free(proc.Mp())
+		}
+	}
+}
+
+func (ctr *container) clean() {
+	if ctr.intHashMap != nil {
+		ctr.intHashMap.Free()
+		ctr.intHashMap = nil
+	}
+	if ctr.strHashMap != nil {
+		ctr.strHashMap.Free()
+		ctr.strHashMap = nil
+	}
+}
+
+func (ctr *container) cleanBatch(proc *process.Process) {
+	if ctr.bat != nil {
+		ctr.bat.Clean(proc.Mp())
+		ctr.bat = nil
+	}
+}
diff --git a/pkg/sql/colexec/group/types.go b/pkg/sql/colexec/group/types.go
index 3c51d48ab..e210f1529 100644
--- a/pkg/sql/colexec/group/types.go
+++ b/pkg/sql/colexec/group/types.go
@@ -16,13 +16,11 @@ package group
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/agg"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -60,50 +58,3 @@ type Argument struct {
 	Types    []types.Type
 	Aggs     []agg.Aggregate // aggregations
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-		ctr.cleanHashMap()
-		ctr.cleanAggVectors(mp)
-		ctr.cleanGroupVectors(mp)
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) cleanAggVectors(mp *mpool.MPool) {
-	for i := range ctr.aggVecs {
-		if ctr.aggVecs[i].needFree && ctr.aggVecs[i].vec != nil {
-			ctr.aggVecs[i].vec.Free(mp)
-			ctr.aggVecs[i].vec = nil
-		}
-	}
-}
-
-func (ctr *container) cleanGroupVectors(mp *mpool.MPool) {
-	for i := range ctr.groupVecs {
-		if ctr.groupVecs[i].needFree && ctr.groupVecs[i].vec != nil {
-			ctr.groupVecs[i].vec.Free(mp)
-			ctr.groupVecs[i].vec = nil
-		}
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.intHashMap != nil {
-		ctr.intHashMap.Free()
-		ctr.intHashMap = nil
-	}
-	if ctr.strHashMap != nil {
-		ctr.strHashMap.Free()
-		ctr.strHashMap = nil
-	}
-}
diff --git a/pkg/sql/colexec/hashbuild/build.go b/pkg/sql/colexec/hashbuild/build.go
index b9cefba90..5a8690777 100644
--- a/pkg/sql/colexec/hashbuild/build.go
+++ b/pkg/sql/colexec/hashbuild/build.go
@@ -61,6 +61,7 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 		case Build:
 			if err := ctr.build(ap, proc, anal); err != nil {
 				ctr.state = End
+				ctr.mp.Free()
 				return true, err
 			}
 			ctr.state = End
@@ -69,9 +70,11 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 				if ap.NeedHashMap {
 					ctr.bat.Ht = hashmap.NewJoinMap(ctr.sels, nil, ctr.mp, ctr.hasNull)
 				}
+				proc.SetInputBatch(ctr.bat)
+				ctr.bat = nil
+			} else {
+				proc.SetInputBatch(nil)
 			}
-			proc.SetInputBatch(ctr.bat)
-			ctr.bat = nil
 			return true, nil
 		}
 	}
@@ -92,6 +95,7 @@ func (ctr *container) build(ap *Argument, proc *process.Process, anal process.An
 		anal.Alloc(int64(bat.Size()))
 		if ctr.bat, err = ctr.bat.Append(proc.Mp(), bat); err != nil {
 			bat.Clean(proc.Mp())
+			ctr.bat.Clean(proc.Mp())
 			return err
 		}
 		bat.Clean(proc.Mp())
@@ -102,7 +106,7 @@ func (ctr *container) build(ap *Argument, proc *process.Process, anal process.An
 	if err := ctr.evalJoinCondition(ctr.bat, ap.Conditions, proc); err != nil {
 		return err
 	}
-	defer ctr.cleanEvalVectors(proc.Mp())
+	defer ctr.freeJoinCondition(proc)
 
 	itr := ctr.mp.NewIterator()
 	count := ctr.bat.Length()
@@ -138,6 +142,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	for i, cond := range conds {
 		vec, err := colexec.EvalExpr(bat, proc, cond)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.Mp())
+				}
+			}
 			return err
 		}
 		ctr.vecs[i] = vec
@@ -152,3 +161,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	}
 	return nil
 }
+
+func (ctr *container) freeJoinCondition(proc *process.Process) {
+	for i := range ctr.evecs {
+		if ctr.evecs[i].needFree {
+			ctr.evecs[i].vec.Free(proc.Mp())
+		}
+	}
+}
diff --git a/pkg/sql/colexec/hashbuild/types.go b/pkg/sql/colexec/hashbuild/types.go
index 6e81db603..74eed2381 100644
--- a/pkg/sql/colexec/hashbuild/types.go
+++ b/pkg/sql/colexec/hashbuild/types.go
@@ -16,12 +16,10 @@ package hashbuild
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -59,38 +57,3 @@ type Argument struct {
 	Typs        []types.Type
 	Conditions  []*plan.Expr
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-		ctr.cleanEvalVectors(mp)
-		if !arg.NeedHashMap {
-			ctr.cleanHashMap()
-		}
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) cleanEvalVectors(mp *mpool.MPool) {
-	for i := range ctr.evecs {
-		if ctr.evecs[i].needFree && ctr.evecs[i].vec != nil {
-			ctr.evecs[i].vec.Free(mp)
-			ctr.evecs[i].vec = nil
-		}
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.mp != nil {
-		ctr.mp.Free()
-		ctr.mp = nil
-	}
-}
diff --git a/pkg/sql/colexec/insert/insert.go b/pkg/sql/colexec/insert/insert.go
index 2abd2f8d1..25a42e71d 100644
--- a/pkg/sql/colexec/insert/insert.go
+++ b/pkg/sql/colexec/insert/insert.go
@@ -44,9 +44,6 @@ type Argument struct {
 	ComputeIndexInfos  []*plan.ComputeIndexInfo
 }
 
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
-
 func String(_ any, buf *bytes.Buffer) {
 	buf.WriteString("insert select")
 }
diff --git a/pkg/sql/colexec/intersect/intersect.go b/pkg/sql/colexec/intersect/intersect.go
index 78015315c..71457e1de 100644
--- a/pkg/sql/colexec/intersect/intersect.go
+++ b/pkg/sql/colexec/intersect/intersect.go
@@ -70,6 +70,7 @@ func Call(idx int, proc *process.Process, argument any) (bool, error) {
 			return false, nil
 
 		case end:
+			arg.ctr.freeContainer(proc)
 			proc.SetInputBatch(nil)
 			return true, nil
 		}
@@ -141,6 +142,7 @@ func (c *container) probeHashTable(proc *process.Process, analyze process.Analyz
 		}
 
 		analyze.Input(btc)
+		defer btc.Clean(proc.Mp())
 
 		c.btc = batch.NewWithSize(len(btc.Vecs))
 		for i := range btc.Vecs {
@@ -192,7 +194,6 @@ func (c *container) probeHashTable(proc *process.Process, analyze process.Analyz
 			if insertcnt > 0 {
 				for pos := range btc.Vecs {
 					if err := vector.UnionBatch(c.btc.Vecs[pos], btc.Vecs[pos], int64(i), insertcnt, needInsert, proc.Mp()); err != nil {
-						btc.Clean(proc.Mp())
 						return false, err
 					}
 				}
@@ -201,8 +202,20 @@ func (c *container) probeHashTable(proc *process.Process, analyze process.Analyz
 
 		analyze.Output(c.btc)
 		proc.SetInputBatch(c.btc)
-		c.btc = nil
-		btc.Clean(proc.Mp())
 		return false, nil
 	}
 }
+
+func (c *container) freeContainer(proc *process.Process) {
+	if c.cnts != nil {
+		for i := range c.cnts {
+			proc.Mp().PutSels(c.cnts[i])
+		}
+		c.cnts = nil
+	}
+
+	if c.hashTable != nil {
+		c.hashTable.Free()
+		c.hashTable = nil
+	}
+}
diff --git a/pkg/sql/colexec/intersect/types.go b/pkg/sql/colexec/intersect/types.go
index 857f1fe84..c24c33541 100644
--- a/pkg/sql/colexec/intersect/types.go
+++ b/pkg/sql/colexec/intersect/types.go
@@ -17,7 +17,6 @@ package intersect
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -50,15 +49,3 @@ type container struct {
 	// process bucket mark
 	inBuckets []uint8
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr.hashTable != nil {
-		ctr.hashTable.Free()
-		ctr.hashTable = nil
-	}
-	if ctr.btc != nil {
-		ctr.btc.Clean(proc.Mp())
-		ctr.btc = nil
-	}
-}
diff --git a/pkg/sql/colexec/intersectall/intersectall.go b/pkg/sql/colexec/intersectall/intersectall.go
index cc47a63f0..2022d115d 100644
--- a/pkg/sql/colexec/intersectall/intersectall.go
+++ b/pkg/sql/colexec/intersectall/intersectall.go
@@ -62,6 +62,8 @@ func Call(idx int, proc *process.Process, argument any) (bool, error) {
 		switch arg.ctr.state {
 		case Build:
 			if err = arg.ctr.build(proc, analyzer); err != nil {
+				arg.ctr.hashTable.Free()
+				arg.ctr.hashTable = nil
 				arg.ctr.state = End
 				return true, err
 			}
@@ -70,6 +72,8 @@ func Call(idx int, proc *process.Process, argument any) (bool, error) {
 			last := false
 			last, err = arg.ctr.probe(proc, analyzer)
 			if err != nil {
+				arg.ctr.hashTable.Free()
+				arg.ctr.hashTable = nil
 				arg.ctr.state = End
 				return true, err
 			}
@@ -79,6 +83,10 @@ func Call(idx int, proc *process.Process, argument any) (bool, error) {
 			}
 			return false, nil
 		case End:
+			if arg.ctr.hashTable != nil {
+				arg.ctr.hashTable.Free()
+				arg.ctr.hashTable = nil
+			}
 			proc.SetInputBatch(nil)
 			return true, nil
 		}
diff --git a/pkg/sql/colexec/intersectall/types.go b/pkg/sql/colexec/intersectall/types.go
index f9537e694..813cffb0c 100644
--- a/pkg/sql/colexec/intersectall/types.go
+++ b/pkg/sql/colexec/intersectall/types.go
@@ -16,7 +16,6 @@ package intersectall
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 type container struct {
@@ -44,17 +43,3 @@ type Argument struct {
 	// buckets count
 	NBucket uint64
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		ctr.cleanHashMap()
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.hashTable != nil {
-		ctr.hashTable.Free()
-		ctr.hashTable = nil
-	}
-}
diff --git a/pkg/sql/colexec/join/join.go b/pkg/sql/colexec/join/join.go
index 837f71721..9c698085f 100644
--- a/pkg/sql/colexec/join/join.go
+++ b/pkg/sql/colexec/join/join.go
@@ -49,6 +49,7 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 		case Build:
 			if err := ctr.build(ap, proc, anal); err != nil {
 				ctr.state = End
+				ctr.mp.Free()
 				return true, err
 			}
 			ctr.state = Probe
@@ -56,6 +57,10 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				ctr.mp.Free()
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -67,6 +72,8 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			}
 			if err := ctr.probe(bat, ap, proc, anal); err != nil {
 				ctr.state = End
+				ctr.mp.Free()
+				proc.SetInputBatch(nil)
 				return true, err
 			}
 			return false, nil
@@ -79,10 +86,8 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 
 func (ctr *container) build(ap *Argument, proc *process.Process, anal process.Analyze) error {
 	bat := <-proc.Reg.MergeReceivers[1].Ch
-	if bat != nil {
-		ctr.bat = bat
-		ctr.mp = bat.Ht.(*hashmap.JoinMap).Dup()
-	}
+	ctr.bat = bat
+	ctr.mp = bat.Ht.(*hashmap.JoinMap).Dup()
 	return nil
 }
 
@@ -101,7 +106,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 	if err := ctr.evalJoinCondition(bat, ap.Conditions[0], proc); err != nil {
 		return err
 	}
-	defer ctr.cleanEvalVectors(proc.Mp())
+	defer ctr.freeJoinCondition(proc)
 	count := bat.Length()
 	mSels := ctr.mp.Sels()
 	itr := ctr.mp.Map().NewIterator()
@@ -157,6 +162,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	for i, cond := range conds {
 		vec, err := colexec.EvalExpr(bat, proc, cond)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.Mp())
+				}
+			}
 			return err
 		}
 		ctr.vecs[i] = vec
@@ -171,3 +181,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	}
 	return nil
 }
+
+func (ctr *container) freeJoinCondition(proc *process.Process) {
+	for i := range ctr.evecs {
+		if ctr.evecs[i].needFree {
+			ctr.evecs[i].vec.Free(proc.Mp())
+		}
+	}
+}
diff --git a/pkg/sql/colexec/join/types.go b/pkg/sql/colexec/join/types.go
index 4d190eef8..1bc79b938 100644
--- a/pkg/sql/colexec/join/types.go
+++ b/pkg/sql/colexec/join/types.go
@@ -16,13 +16,11 @@ package join
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -58,36 +56,3 @@ type Argument struct {
 	Cond       *plan.Expr
 	Conditions [][]*plan.Expr
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-		ctr.cleanEvalVectors(mp)
-		ctr.cleanHashMap()
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.mp != nil {
-		ctr.mp.Free()
-		ctr.mp = nil
-	}
-}
-
-func (ctr *container) cleanEvalVectors(mp *mpool.MPool) {
-	for i := range ctr.evecs {
-		if ctr.evecs[i].needFree && ctr.evecs[i].vec != nil {
-			ctr.evecs[i].vec.Free(mp)
-			ctr.evecs[i].vec = nil
-		}
-	}
-}
diff --git a/pkg/sql/colexec/left/join.go b/pkg/sql/colexec/left/join.go
index c8c799954..fa5801a6d 100644
--- a/pkg/sql/colexec/left/join.go
+++ b/pkg/sql/colexec/left/join.go
@@ -54,6 +54,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 		case Build:
 			if err := ctr.build(ap, proc, anal); err != nil {
 				ctr.state = End
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
 				return true, err
 			}
 			ctr.state = Probe
@@ -61,6 +64,12 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -69,12 +78,20 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			if ctr.bat.Length() == 0 {
 				if err := ctr.emptyProbe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 
 			} else {
 				if err := ctr.probe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 			}
@@ -130,7 +147,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 	if err := ctr.evalJoinCondition(bat, ap.Conditions[0], proc); err != nil {
 		return err
 	}
-	defer ctr.cleanEvalVectors(proc.Mp())
+	defer ctr.freeJoinCondition(proc)
 	count := bat.Length()
 	mSels := ctr.mp.Sels()
 	itr := ctr.mp.Map().NewIterator()
@@ -222,6 +239,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	for i, cond := range conds {
 		vec, err := colexec.EvalExpr(bat, proc, cond)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.Mp())
+				}
+			}
 			return err
 		}
 		ctr.vecs[i] = vec
@@ -236,3 +258,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	}
 	return nil
 }
+
+func (ctr *container) freeJoinCondition(proc *process.Process) {
+	for i := range ctr.evecs {
+		if ctr.evecs[i].needFree {
+			ctr.evecs[i].vec.Free(proc.Mp())
+		}
+	}
+}
diff --git a/pkg/sql/colexec/left/types.go b/pkg/sql/colexec/left/types.go
index 2a3502e2b..7a52519c6 100644
--- a/pkg/sql/colexec/left/types.go
+++ b/pkg/sql/colexec/left/types.go
@@ -16,13 +16,11 @@ package left
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -58,36 +56,3 @@ type Argument struct {
 	Cond       *plan.Expr
 	Conditions [][]*plan.Expr
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-		ctr.cleanEvalVectors(mp)
-		ctr.cleanHashMap()
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.mp != nil {
-		ctr.mp.Free()
-		ctr.mp = nil
-	}
-}
-
-func (ctr *container) cleanEvalVectors(mp *mpool.MPool) {
-	for i := range ctr.evecs {
-		if ctr.evecs[i].needFree && ctr.evecs[i].vec != nil {
-			ctr.evecs[i].vec.Free(mp)
-			ctr.evecs[i].vec = nil
-		}
-	}
-}
diff --git a/pkg/sql/colexec/limit/types.go b/pkg/sql/colexec/limit/types.go
index 50c5b5c4a..ef71c274a 100644
--- a/pkg/sql/colexec/limit/types.go
+++ b/pkg/sql/colexec/limit/types.go
@@ -14,12 +14,7 @@
 
 package limit
 
-import "github.com/matrixorigin/matrixone/pkg/vm/process"
-
 type Argument struct {
 	Seen  uint64 // seen is the number of tuples seen so far
 	Limit uint64
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/loopanti/join.go b/pkg/sql/colexec/loopanti/join.go
index bc55b68c3..0ca5e4a9f 100644
--- a/pkg/sql/colexec/loopanti/join.go
+++ b/pkg/sql/colexec/loopanti/join.go
@@ -52,6 +52,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -60,11 +63,13 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			if ctr.bat == nil || ctr.bat.Length() == 0 {
 				if err := ctr.emptyProbe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 			} else {
 				if err := ctr.probe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 			}
@@ -121,17 +126,16 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 				break
 			}
 		}
+		defer vec.Free(proc.Mp())
 		if !matched && !nulls.Any(vec.Nsp) {
 			for k, pos := range ap.Result {
 				if err := vector.UnionOne(rbat.Vecs[k], bat.Vecs[pos], int64(i), proc.Mp()); err != nil {
 					rbat.Clean(proc.Mp())
-					vec.Free(proc.Mp())
 					return err
 				}
 			}
 			rbat.Zs = append(rbat.Zs, bat.Zs[i])
 		}
-		vec.Free(proc.Mp())
 	}
 	rbat.ExpandNulls()
 	anal.Output(rbat)
diff --git a/pkg/sql/colexec/loopanti/types.go b/pkg/sql/colexec/loopanti/types.go
index 78e778d11..a495d7853 100644
--- a/pkg/sql/colexec/loopanti/types.go
+++ b/pkg/sql/colexec/loopanti/types.go
@@ -15,11 +15,9 @@
 package loopanti
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -39,17 +37,3 @@ type Argument struct {
 	Cond   *plan.Expr
 	Typs   []types.Type
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		ctr.cleanBatch(proc.Mp())
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
diff --git a/pkg/sql/colexec/loopjoin/join.go b/pkg/sql/colexec/loopjoin/join.go
index 4d023ba7d..6bd8b8ec8 100644
--- a/pkg/sql/colexec/loopjoin/join.go
+++ b/pkg/sql/colexec/loopjoin/join.go
@@ -51,6 +51,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -62,6 +65,7 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			}
 			if err := ctr.probe(bat, ap, proc, anal); err != nil {
 				ctr.state = End
+				proc.SetInputBatch(nil)
 				return true, err
 			}
 			return false, nil
@@ -98,7 +102,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 		if err != nil {
 			return err
 		}
-		bs := vector.MustTCols[bool](vec)
+		bs := vec.Col.([]bool)
 		if len(bs) == 1 {
 			if bs[0] {
 				for j := 0; j < len(ctr.bat.Zs); j++ {
@@ -106,13 +110,11 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 						if rp.Rel == 0 {
 							if err := vector.UnionOne(rbat.Vecs[k], bat.Vecs[rp.Pos], int64(i), proc.Mp()); err != nil {
 								rbat.Clean(proc.Mp())
-								vec.Free(proc.Mp())
 								return err
 							}
 						} else {
 							if err := vector.UnionOne(rbat.Vecs[k], ctr.bat.Vecs[rp.Pos], int64(j), proc.Mp()); err != nil {
 								rbat.Clean(proc.Mp())
-								vec.Free(proc.Mp())
 								return err
 							}
 						}
@@ -127,13 +129,11 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 						if rp.Rel == 0 {
 							if err := vector.UnionOne(rbat.Vecs[k], bat.Vecs[rp.Pos], int64(i), proc.Mp()); err != nil {
 								rbat.Clean(proc.Mp())
-								vec.Free(proc.Mp())
 								return err
 							}
 						} else {
 							if err := vector.UnionOne(rbat.Vecs[k], ctr.bat.Vecs[rp.Pos], int64(j), proc.Mp()); err != nil {
 								rbat.Clean(proc.Mp())
-								vec.Free(proc.Mp())
 								return err
 							}
 						}
@@ -142,7 +142,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 				}
 			}
 		}
-		vec.Free(proc.Mp())
+		vector.Clean(vec, proc.Mp())
 	}
 	rbat.ExpandNulls()
 	anal.Output(rbat)
diff --git a/pkg/sql/colexec/loopjoin/types.go b/pkg/sql/colexec/loopjoin/types.go
index 3ce7acbd7..2f1756896 100644
--- a/pkg/sql/colexec/loopjoin/types.go
+++ b/pkg/sql/colexec/loopjoin/types.go
@@ -15,12 +15,10 @@
 package loopjoin
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -40,17 +38,3 @@ type Argument struct {
 	Result []colexec.ResultPos
 	Typs   []types.Type
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		ctr.cleanBatch(proc.Mp())
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
diff --git a/pkg/sql/colexec/loopleft/join.go b/pkg/sql/colexec/loopleft/join.go
index 05617f936..bd9002d1b 100644
--- a/pkg/sql/colexec/loopleft/join.go
+++ b/pkg/sql/colexec/loopleft/join.go
@@ -56,6 +56,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -64,12 +67,14 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			if ctr.bat.Length() == 0 {
 				if err := ctr.emptyProbe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 
 			} else {
 				if err := ctr.probe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 			}
@@ -128,7 +133,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 		if err != nil {
 			return err
 		}
-		bs := vector.MustTCols[bool](vec)
+		bs := vec.Col.([]bool)
 		if len(bs) == 1 {
 			if bs[0] {
 				for j := 0; j < len(ctr.bat.Zs); j++ {
@@ -137,13 +142,11 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 						if rp.Rel == 0 {
 							if err := vector.UnionOne(rbat.Vecs[k], bat.Vecs[rp.Pos], int64(i), proc.Mp()); err != nil {
 								rbat.Clean(proc.Mp())
-								vec.Free(proc.Mp())
 								return err
 							}
 						} else {
 							if err := vector.UnionOne(rbat.Vecs[k], ctr.bat.Vecs[rp.Pos], int64(j), proc.Mp()); err != nil {
 								rbat.Clean(proc.Mp())
-								vec.Free(proc.Mp())
 								return err
 							}
 						}
@@ -160,13 +163,11 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 						if rp.Rel == 0 {
 							if err := vector.UnionOne(rbat.Vecs[k], bat.Vecs[rp.Pos], int64(i), proc.Mp()); err != nil {
 								rbat.Clean(proc.Mp())
-								vec.Free(proc.Mp())
 								return err
 							}
 						} else {
 							if err := vector.UnionOne(rbat.Vecs[k], ctr.bat.Vecs[rp.Pos], int64(j), proc.Mp()); err != nil {
 								rbat.Clean(proc.Mp())
-								vec.Free(proc.Mp())
 								return err
 							}
 						}
@@ -175,7 +176,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 				}
 			}
 		}
-		vec.Free(proc.Mp())
+		vector.Clean(vec, proc.Mp())
 		if !matched {
 			for k, rp := range ap.Result {
 				if rp.Rel == 0 {
diff --git a/pkg/sql/colexec/loopleft/types.go b/pkg/sql/colexec/loopleft/types.go
index 64e10caa8..85aab10d6 100644
--- a/pkg/sql/colexec/loopleft/types.go
+++ b/pkg/sql/colexec/loopleft/types.go
@@ -15,12 +15,10 @@
 package loopleft
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -40,17 +38,3 @@ type Argument struct {
 	Cond   *plan.Expr
 	Result []colexec.ResultPos
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		ctr.cleanBatch(proc.Mp())
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
diff --git a/pkg/sql/colexec/loopsemi/join.go b/pkg/sql/colexec/loopsemi/join.go
index 1543669f9..2e3fbcfbe 100644
--- a/pkg/sql/colexec/loopsemi/join.go
+++ b/pkg/sql/colexec/loopsemi/join.go
@@ -51,6 +51,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -62,6 +65,7 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			}
 			if err := ctr.probe(bat, ap, proc, anal); err != nil {
 				ctr.state = End
+				proc.SetInputBatch(nil)
 				return true, err
 			}
 			return false, nil
@@ -94,21 +98,21 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 		if err != nil {
 			return err
 		}
-		bs := vector.MustTCols[bool](vec)
+		bs := vec.Col.([]bool)
 		for _, b := range bs {
 			if b {
 				for k, pos := range ap.Result {
 					if err := vector.UnionOne(rbat.Vecs[k], bat.Vecs[pos], int64(i), proc.Mp()); err != nil {
 						rbat.Clean(proc.Mp())
-						vec.Free(proc.Mp())
 						return err
 					}
 				}
+				vector.Clean(vec, proc.Mp())
 				rbat.Zs = append(rbat.Zs, bat.Zs[i])
 				break
 			}
 		}
-		vec.Free(proc.Mp())
+		vector.Clean(vec, proc.Mp())
 	}
 	rbat.ExpandNulls()
 	anal.Output(rbat)
diff --git a/pkg/sql/colexec/loopsemi/types.go b/pkg/sql/colexec/loopsemi/types.go
index 56be58367..4acce2f24 100644
--- a/pkg/sql/colexec/loopsemi/types.go
+++ b/pkg/sql/colexec/loopsemi/types.go
@@ -15,11 +15,9 @@
 package loopsemi
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -39,17 +37,3 @@ type Argument struct {
 	Cond   *plan.Expr
 	Typs   []types.Type
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		ctr.cleanBatch(proc.Mp())
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
diff --git a/pkg/sql/colexec/loopsingle/join.go b/pkg/sql/colexec/loopsingle/join.go
index 4f23a1c03..aa29ee8bc 100644
--- a/pkg/sql/colexec/loopsingle/join.go
+++ b/pkg/sql/colexec/loopsingle/join.go
@@ -57,6 +57,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -65,12 +68,14 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			if ctr.bat.Length() == 0 {
 				if err := ctr.emptyProbe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 
 			} else {
 				if err := ctr.probe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 			}
@@ -126,11 +131,11 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 		if err != nil {
 			return err
 		}
-		bs := vector.MustTCols[bool](vec)
+		defer vector.Clean(vec, proc.Mp())
+		bs := vec.Col.([]bool)
 		if len(bs) == 1 {
 			if bs[0] {
 				if len(ctr.bat.Zs) > 1 {
-					vec.Free(proc.Mp())
 					return moerr.NewInternalError("scalar subquery returns more than 1 row")
 				}
 				unmatched = false
@@ -138,7 +143,6 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 					if rp.Rel != 0 {
 						if err := vector.UnionOne(rbat.Vecs[k], ctr.bat.Vecs[rp.Pos], 0, proc.Mp()); err != nil {
 							rbat.Clean(proc.Mp())
-							vec.Free(proc.Mp())
 							return err
 						}
 					}
@@ -148,7 +152,6 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 			for j, b := range bs {
 				if b {
 					if !unmatched {
-						vec.Free(proc.Mp())
 						return moerr.NewInternalError("scalar subquery returns more than 1 row")
 					}
 					unmatched = false
@@ -156,7 +159,6 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 						if rp.Rel != 0 {
 							if err := vector.UnionOne(rbat.Vecs[k], ctr.bat.Vecs[rp.Pos], int64(j), proc.Mp()); err != nil {
 								rbat.Clean(proc.Mp())
-								vec.Free(proc.Mp())
 								return err
 							}
 						}
@@ -169,13 +171,11 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 				if rp.Rel != 0 {
 					if err := vector.UnionNull(rbat.Vecs[k], ctr.bat.Vecs[rp.Pos], proc.Mp()); err != nil {
 						rbat.Clean(proc.Mp())
-						vec.Free(proc.Mp())
 						return err
 					}
 				}
 			}
 		}
-		vec.Free(proc.Mp())
 	}
 	for i, rp := range ap.Result {
 		if rp.Rel == 0 {
diff --git a/pkg/sql/colexec/loopsingle/types.go b/pkg/sql/colexec/loopsingle/types.go
index 282c43d8d..a20b15106 100644
--- a/pkg/sql/colexec/loopsingle/types.go
+++ b/pkg/sql/colexec/loopsingle/types.go
@@ -15,12 +15,10 @@
 package loopsingle
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -40,17 +38,3 @@ type Argument struct {
 	Typs   []types.Type
 	Result []colexec.ResultPos
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		ctr.cleanBatch(proc.Mp())
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
diff --git a/pkg/sql/colexec/mark/join.go b/pkg/sql/colexec/mark/join.go
index 1aacbe926..7a276b9eb 100644
--- a/pkg/sql/colexec/mark/join.go
+++ b/pkg/sql/colexec/mark/join.go
@@ -76,6 +76,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 		case Build:
 			if err := ctr.build(ap, proc, anal); err != nil {
 				ctr.state = End
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
 				return true, err
 			}
 			ctr.state = Probe
@@ -83,6 +86,16 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				ctr.freeBuildEqVec(proc)
+				if ctr.nullWithBatch != nil {
+					ctr.nullWithBatch.Clean(proc.Mp())
+				}
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -91,11 +104,26 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			if ctr.bat == nil || ctr.bat.Length() == 0 {
 				if err := ctr.emptyProbe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 			} else {
 				if err := ctr.probe(bat, ap, proc, anal); err != nil {
+					ctr.freeBuildEqVec(proc)
+					if ctr.nullWithBatch != nil {
+						ctr.nullWithBatch.Clean(proc.Mp())
+					}
 					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
+					if ctr.bat != nil {
+						ctr.bat.Clean(proc.Mp())
+					}
+					proc.SetInputBatch(nil)
 					return true, err
 				}
 			}
@@ -182,7 +210,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 	if err := ctr.evalJoinProbeCondition(bat, ap.Conditions[0], proc); err != nil {
 		return err
 	}
-	defer ctr.cleanEvalVectors(proc.Mp())
+	defer ctr.freeProbeEqVec(proc)
 	count := bat.Length()
 	itr := ctr.mp.Map().NewIterator()
 	mSels := ctr.mp.Sels()
@@ -277,6 +305,11 @@ func (ctr *container) evalJoinProbeCondition(bat *batch.Batch, conds []*plan.Exp
 	for i, cond := range conds {
 		vec, err := colexec.EvalExpr(bat, proc, cond)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.Mp())
+				}
+			}
 			return err
 		}
 		ctr.vecs[i] = vec
@@ -297,6 +330,11 @@ func (ctr *container) evalJoinBuildCondition(bat *batch.Batch, conds []*plan.Exp
 	for i, cond := range conds {
 		vec, err := colexec.EvalExpr(bat, proc, cond)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.Mp())
+				}
+			}
 			return err
 		}
 		ctr.buildEqVec[i] = vec
@@ -312,6 +350,22 @@ func (ctr *container) evalJoinBuildCondition(bat *batch.Batch, conds []*plan.Exp
 	return nil
 }
 
+func (ctr *container) freeProbeEqVec(proc *process.Process) {
+	for i := range ctr.evecs {
+		if ctr.evecs[i].needFree {
+			ctr.evecs[i].vec.Free(proc.Mp())
+		}
+	}
+}
+
+func (ctr *container) freeBuildEqVec(proc *process.Process) {
+	for i := range ctr.buildEqEvecs {
+		if ctr.buildEqEvecs[i].needFree {
+			ctr.buildEqEvecs[i].vec.Free(proc.Mp())
+		}
+	}
+}
+
 // calculate the state of non-equal conditions for those tuples in JoinMap
 func (ctr *container) nonEqJoinInMap(ap *Argument, mSels [][]int64, vals []uint64, k int, i int, proc *process.Process, bat *batch.Batch) (resultType, error) {
 	if ap.Cond != nil {
@@ -325,7 +379,7 @@ func (ctr *container) nonEqJoinInMap(ap *Argument, mSels [][]int64, vals []uint6
 			if vec.Nsp.Contains(0) {
 				condState = condUnkown
 			}
-			bs := vector.MustTCols[bool](vec)
+			bs := vec.Col.([]bool)
 			if bs[0] {
 				condState = condTrue
 				vec.Free(proc.Mp())
diff --git a/pkg/sql/colexec/mark/types.go b/pkg/sql/colexec/mark/types.go
index de835d1ed..5b8f3682f 100644
--- a/pkg/sql/colexec/mark/types.go
+++ b/pkg/sql/colexec/mark/types.go
@@ -16,13 +16,11 @@ package mark
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/nulls"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -162,50 +160,3 @@ type Argument struct {
 
 	OnList []*plan.Expr
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-		ctr.cleanEvalVectors(mp)
-		ctr.cleanEqVectors(mp)
-		ctr.cleanHashMap()
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-	if ctr.nullWithBatch != nil {
-		ctr.nullWithBatch.Clean(mp)
-		ctr.nullWithBatch = nil
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.mp != nil {
-		ctr.mp.Free()
-		ctr.mp = nil
-	}
-}
-
-func (ctr *container) cleanEvalVectors(mp *mpool.MPool) {
-	for i := range ctr.evecs {
-		if ctr.evecs[i].needFree && ctr.evecs[i].vec != nil {
-			ctr.evecs[i].vec.Free(mp)
-			ctr.evecs[i].vec = nil
-		}
-	}
-}
-
-func (ctr *container) cleanEqVectors(mp *mpool.MPool) {
-	for i := range ctr.buildEqEvecs {
-		if ctr.buildEqEvecs[i].needFree && ctr.buildEqEvecs[i].vec != nil {
-			ctr.buildEqEvecs[i].vec.Free(mp)
-			ctr.buildEqEvecs[i].vec = nil
-		}
-	}
-}
diff --git a/pkg/sql/colexec/merge/merge.go b/pkg/sql/colexec/merge/merge.go
index a45e833db..a76892d43 100644
--- a/pkg/sql/colexec/merge/merge.go
+++ b/pkg/sql/colexec/merge/merge.go
@@ -17,6 +17,7 @@ package merge
 import (
 	"bytes"
 
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
@@ -40,7 +41,13 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			proc.SetInputBatch(nil)
 			return true, nil
 		}
-		bat := <-proc.Reg.MergeReceivers[ap.ctr.i].Ch
+		reg := proc.Reg.MergeReceivers[ap.ctr.i]
+		var bat *batch.Batch = nil
+		select {
+		case <-reg.Ctx.Done():
+		case bat = <-reg.Ch:
+		}
+
 		if bat == nil {
 			proc.Reg.MergeReceivers = append(proc.Reg.MergeReceivers[:ap.ctr.i], proc.Reg.MergeReceivers[ap.ctr.i+1:]...)
 			if ap.ctr.i >= len(proc.Reg.MergeReceivers) {
diff --git a/pkg/sql/colexec/merge/types.go b/pkg/sql/colexec/merge/types.go
index 7e73aa93b..0bb7fb8a8 100644
--- a/pkg/sql/colexec/merge/types.go
+++ b/pkg/sql/colexec/merge/types.go
@@ -14,8 +14,6 @@
 
 package merge
 
-import "github.com/matrixorigin/matrixone/pkg/vm/process"
-
 type container struct {
 	i int
 }
@@ -23,6 +21,3 @@ type container struct {
 type Argument struct {
 	ctr *container
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/mergegroup/group.go b/pkg/sql/colexec/mergegroup/group.go
index 4b97f71bd..2d8fd0cfb 100644
--- a/pkg/sql/colexec/mergegroup/group.go
+++ b/pkg/sql/colexec/mergegroup/group.go
@@ -50,12 +50,15 @@ func Call(idx int, proc *process.Process, arg interface{}) (bool, error) {
 			}
 			ctr.state = Eval
 		case Eval:
+			ctr.state = End
 			if ctr.bat != nil {
 				if ap.NeedEval {
 					for i, agg := range ctr.bat.Aggs {
 						vec, err := agg.Eval(proc.Mp())
 						if err != nil {
 							ctr.state = End
+							ctr.clean()
+							ctr.bat.Clean(proc.Mp())
 							return false, err
 						}
 						ctr.bat.Aggs[i] = nil
@@ -69,10 +72,10 @@ func Call(idx int, proc *process.Process, arg interface{}) (bool, error) {
 				anal.Output(ctr.bat)
 				ctr.bat.ExpandNulls()
 			}
+			ctr.clean()
 			proc.SetInputBatch(ctr.bat)
 			ctr.bat = nil
-			ctr.state = End
-			return false, nil
+			return true, nil
 		case End:
 			proc.SetInputBatch(nil)
 			return true, nil
@@ -157,6 +160,8 @@ func (ctr *container) process(bat *batch.Batch, proc *process.Process) error {
 		err = ctr.processHStr(bat, proc)
 	}
 	if err != nil {
+		ctr.clean()
+		ctr.cleanBatch(proc)
 		return err
 	}
 	return nil
@@ -267,3 +272,21 @@ func (ctr *container) batchFill(i int, n int, bat *batch.Batch, vals []uint64, h
 	}
 	return nil
 }
+
+func (ctr *container) clean() {
+	if ctr.intHashMap != nil {
+		ctr.intHashMap.Free()
+		ctr.intHashMap = nil
+	}
+	if ctr.strHashMap != nil {
+		ctr.strHashMap.Free()
+		ctr.strHashMap = nil
+	}
+}
+
+func (ctr *container) cleanBatch(proc *process.Process) {
+	if ctr.bat != nil {
+		ctr.bat.Clean(proc.Mp())
+		ctr.bat = nil
+	}
+}
diff --git a/pkg/sql/colexec/mergegroup/types.go b/pkg/sql/colexec/mergegroup/types.go
index 0c91acc46..9d84398bc 100644
--- a/pkg/sql/colexec/mergegroup/types.go
+++ b/pkg/sql/colexec/mergegroup/types.go
@@ -16,9 +16,7 @@ package mergegroup
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -49,30 +47,3 @@ type Argument struct {
 	NeedEval bool // need to projection the aggregate column
 	ctr      *container
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-		ctr.cleanHashMap()
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.intHashMap != nil {
-		ctr.intHashMap.Free()
-		ctr.intHashMap = nil
-	}
-	if ctr.strHashMap != nil {
-		ctr.strHashMap.Free()
-		ctr.strHashMap = nil
-	}
-}
diff --git a/pkg/sql/colexec/mergelimit/limit.go b/pkg/sql/colexec/mergelimit/limit.go
index 0c7d6d63f..dc26af4de 100644
--- a/pkg/sql/colexec/mergelimit/limit.go
+++ b/pkg/sql/colexec/mergelimit/limit.go
@@ -38,9 +38,23 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 	ap := arg.(*Argument)
 	anal := proc.GetAnalyze(idx)
 	anal.Start()
-	ok, err := ap.ctr.eval(ap, proc, anal)
-	anal.Stop()
-	return ok, err
+	defer anal.Stop()
+	for {
+		switch ap.ctr.state {
+		case Eval:
+			ok, err := ap.ctr.eval(ap, proc, anal)
+			if err != nil {
+				return ok, err
+			}
+			if ok {
+				ap.ctr.state = End
+			}
+			return ok, err
+		default:
+			proc.SetInputBatch(nil)
+			return true, nil
+		}
+	}
 }
 
 func (ctr *container) eval(ap *Argument, proc *process.Process, anal process.Analyze) (bool, error) {
@@ -63,7 +77,7 @@ func (ctr *container) eval(ap *Argument, proc *process.Process, anal process.Ana
 		if ap.ctr.seen >= ap.Limit {
 			proc.SetInputBatch(nil)
 			bat.Clean(proc.Mp())
-			return false, nil
+			return true, nil
 		}
 		newSeen := ap.ctr.seen + uint64(bat.Length())
 		if newSeen < ap.Limit {
@@ -80,7 +94,6 @@ func (ctr *container) eval(ap *Argument, proc *process.Process, anal process.Ana
 			return false, nil
 		}
 	}
-	proc.SetInputBatch(nil)
 	return true, nil
 
 }
diff --git a/pkg/sql/colexec/mergelimit/types.go b/pkg/sql/colexec/mergelimit/types.go
index a840632d7..b30cd0fff 100644
--- a/pkg/sql/colexec/mergelimit/types.go
+++ b/pkg/sql/colexec/mergelimit/types.go
@@ -14,10 +14,14 @@
 
 package mergelimit
 
-import "github.com/matrixorigin/matrixone/pkg/vm/process"
+const (
+	Eval = iota
+	End
+)
 
 type container struct {
-	seen uint64
+	state int
+	seen  uint64
 }
 
 type Argument struct {
@@ -26,6 +30,3 @@ type Argument struct {
 	// ctr stores the attributes needn't do Serialization work
 	ctr *container
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/mergeoffset/offset.go b/pkg/sql/colexec/mergeoffset/offset.go
index 2b23c972b..65448d8ff 100644
--- a/pkg/sql/colexec/mergeoffset/offset.go
+++ b/pkg/sql/colexec/mergeoffset/offset.go
@@ -35,10 +35,24 @@ func Prepare(_ *process.Process, arg interface{}) error {
 
 func Call(idx int, proc *process.Process, arg interface{}) (bool, error) {
 	ap := arg.(*Argument)
-	anal := proc.GetAnalyze(idx)
-	ok, err := ap.ctr.eval(ap, proc, anal)
-	anal.Stop()
-	return ok, err
+	for {
+		switch ap.ctr.state {
+		case Eval:
+			anal := proc.GetAnalyze(idx)
+			defer anal.Stop()
+			ok, err := ap.ctr.eval(ap, proc, anal)
+			if err != nil {
+				return ok, err
+			}
+			if ok {
+				ap.ctr.state = End
+			}
+			return ok, err
+		default:
+			proc.SetInputBatch(nil)
+			return true, nil
+		}
+	}
 }
 
 func (ctr *container) eval(ap *Argument, proc *process.Process, anal process.Analyze) (bool, error) {
@@ -75,9 +89,9 @@ func (ctr *container) eval(ap *Argument, proc *process.Process, anal process.Ana
 		}
 		ap.ctr.seen += uint64(length)
 		bat.Clean(proc.Mp())
+		proc.SetInputBatch(nil)
 		i--
 	}
-	proc.SetInputBatch(nil)
 	return true, nil
 }
 
diff --git a/pkg/sql/colexec/mergeoffset/types.go b/pkg/sql/colexec/mergeoffset/types.go
index 021f334bb..b1eeb5945 100644
--- a/pkg/sql/colexec/mergeoffset/types.go
+++ b/pkg/sql/colexec/mergeoffset/types.go
@@ -14,10 +14,14 @@
 
 package mergeoffset
 
-import "github.com/matrixorigin/matrixone/pkg/vm/process"
+const (
+	Eval = iota
+	End
+)
 
 type container struct {
-	seen uint64
+	state int
+	seen  uint64
 }
 
 type Argument struct {
@@ -26,6 +30,3 @@ type Argument struct {
 	// ctr contains the attributes needn't do serialization work
 	ctr *container
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/mergeorder/order.go b/pkg/sql/colexec/mergeorder/order.go
index b85981777..fa6db93b7 100644
--- a/pkg/sql/colexec/mergeorder/order.go
+++ b/pkg/sql/colexec/mergeorder/order.go
@@ -143,6 +143,7 @@ func (ctr *container) build(ap *Argument, proc *process.Process, anal process.An
 			} else {
 				if err := ctr.processBatch(bat, proc); err != nil {
 					bat.Clean(proc.Mp())
+					ctr.bat.Clean(proc.Mp())
 					return err
 				}
 				bat.Clean(proc.Mp())
@@ -223,7 +224,7 @@ func (ctr *container) processBatch(bat2 *batch.Batch, proc *process.Process) err
 		}
 		rbat.Zs = append(rbat.Zs, bat2.Zs[j:]...)
 	}
-	ctr.cleanBatch(proc.Mp())
+	ctr.bat.Clean(proc.Mp())
 	ctr.bat = rbat
 	return nil
 }
diff --git a/pkg/sql/colexec/mergeorder/types.go b/pkg/sql/colexec/mergeorder/types.go
index 7dcf2b446..11bb38373 100644
--- a/pkg/sql/colexec/mergeorder/types.go
+++ b/pkg/sql/colexec/mergeorder/types.go
@@ -15,11 +15,9 @@
 package mergeorder
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/compare"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -41,18 +39,3 @@ type Argument struct {
 	ctr *container          // ctr stores the attributes needn't do Serialization work
 	Fs  []*plan.OrderBySpec // Fields store the order information
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
diff --git a/pkg/sql/colexec/mergetop/top.go b/pkg/sql/colexec/mergetop/top.go
index 7ac2cd52d..4f88fae81 100644
--- a/pkg/sql/colexec/mergetop/top.go
+++ b/pkg/sql/colexec/mergetop/top.go
@@ -214,10 +214,11 @@ func (ctr *container) eval(limit int64, proc *process.Process, anal process.Anal
 		sels[len(sels)-1-i] = heap.Pop(ctr).(int64)
 	}
 	if err := ctr.bat.Shuffle(sels, proc.Mp()); err != nil {
-		return err
+		ctr.bat.Clean(proc.Mp())
+		ctr.bat = nil
 	}
 	for i := ctr.n; i < len(ctr.bat.Vecs); i++ {
-		ctr.bat.Vecs[i].Free(proc.Mp())
+		vector.Clean(ctr.bat.Vecs[i], proc.Mp())
 	}
 	ctr.bat.Vecs = ctr.bat.Vecs[:ctr.n]
 	ctr.bat.ExpandNulls()
diff --git a/pkg/sql/colexec/mergetop/types.go b/pkg/sql/colexec/mergetop/types.go
index 9517f5ef1..5f00dc9b4 100644
--- a/pkg/sql/colexec/mergetop/types.go
+++ b/pkg/sql/colexec/mergetop/types.go
@@ -15,11 +15,9 @@
 package mergetop
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/compare"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -44,21 +42,6 @@ type Argument struct {
 	Fs    []*plan.OrderBySpec // Fs store the order information
 }
 
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
 func (ctr *container) compare(vi, vj int, i, j int64) int {
 	for _, pos := range ctr.poses {
 		if r := ctr.cmps[pos].Compare(vi, vj, i, j); r != 0 {
diff --git a/pkg/sql/colexec/minus/minus.go b/pkg/sql/colexec/minus/minus.go
index a1992ab9c..0432956dc 100644
--- a/pkg/sql/colexec/minus/minus.go
+++ b/pkg/sql/colexec/minus/minus.go
@@ -58,6 +58,7 @@ func Call(idx int, proc *process.Process, argument any) (bool, error) {
 		case buildingHashMap:
 			// step 1: build the hash table by all right batches.
 			if err = arg.ctr.buildHashTable(proc, analyze, 1); err != nil {
+				arg.ctr.hashTable.Free()
 				arg.ctr.state = operatorEnd
 				return true, err
 			}
@@ -70,6 +71,8 @@ func Call(idx int, proc *process.Process, argument any) (bool, error) {
 			last := false
 			last, err = arg.ctr.probeHashTable(proc, analyze, 0)
 			if err != nil {
+				arg.ctr.bat.Clean(proc.Mp())
+				arg.ctr.hashTable.Free()
 				arg.ctr.state = operatorEnd
 				return true, err
 			}
@@ -80,6 +83,10 @@ func Call(idx int, proc *process.Process, argument any) (bool, error) {
 			return false, nil
 		case operatorEnd:
 			// operator over.
+			if arg.ctr.hashTable != nil {
+				arg.ctr.hashTable.Free()
+				arg.ctr.hashTable = nil
+			}
 			proc.SetInputBatch(nil)
 			return true, nil
 		}
@@ -181,7 +188,6 @@ func (ctr *container) probeHashTable(proc *process.Process, ana process.Analyze,
 		}
 		ana.Output(ctr.bat)
 		proc.SetInputBatch(ctr.bat)
-		ctr.bat = nil
 		bat.Clean(proc.Mp())
 		return false, nil
 	}
diff --git a/pkg/sql/colexec/minus/types.go b/pkg/sql/colexec/minus/types.go
index 0ca379bdd..1f60e7d4a 100644
--- a/pkg/sql/colexec/minus/types.go
+++ b/pkg/sql/colexec/minus/types.go
@@ -16,9 +16,7 @@ package minus
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -44,23 +42,3 @@ type container struct {
 	// result batch of minus column execute operator
 	bat *batch.Batch
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	mp := proc.Mp()
-	arg.ctr.cleanBatch(mp)
-	arg.ctr.cleanHashMap()
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.hashTable != nil {
-		ctr.hashTable.Free()
-		ctr.hashTable = nil
-	}
-}
diff --git a/pkg/sql/colexec/offset/types.go b/pkg/sql/colexec/offset/types.go
index 3c495c31a..d8dfb8d82 100644
--- a/pkg/sql/colexec/offset/types.go
+++ b/pkg/sql/colexec/offset/types.go
@@ -14,12 +14,7 @@
 
 package offset
 
-import "github.com/matrixorigin/matrixone/pkg/vm/process"
-
 type Argument struct {
 	Seen   uint64 // seen is the number of tuples seen so far
 	Offset uint64
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/order/order.go b/pkg/sql/colexec/order/order.go
index c1d55c14f..a6b19c258 100644
--- a/pkg/sql/colexec/order/order.go
+++ b/pkg/sql/colexec/order/order.go
@@ -41,7 +41,7 @@ func String(arg any, buf *bytes.Buffer) {
 
 func Prepare(_ *process.Process, arg any) error {
 	ap := arg.(*Argument)
-	ap.ctr = new(container)
+	ap.ctr = new(Container)
 	{
 		ap.ctr.desc = make([]bool, len(ap.Fs))
 		ap.ctr.nullsLast = make([]bool, len(ap.Fs))
@@ -75,7 +75,7 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 	return ap.ctr.process(ap, bat, proc)
 }
 
-func (ctr *container) process(ap *Argument, bat *batch.Batch, proc *process.Process) (bool, error) {
+func (ctr *Container) process(ap *Argument, bat *batch.Batch, proc *process.Process) (bool, error) {
 	for i := 0; i < bat.VectorCount(); i++ {
 		vec := bat.GetVector(int32(i))
 		if vec.IsOriginal() {
@@ -90,6 +90,11 @@ func (ctr *container) process(ap *Argument, bat *batch.Batch, proc *process.Proc
 	for i, f := range ap.Fs {
 		vec, err := colexec.EvalExpr(bat, proc, f.Expr)
 		if err != nil {
+			for j := 0; j < i; j++ {
+				if ctr.vecs[j].needFree {
+					vector.Clean(ctr.vecs[j].vec, proc.Mp())
+				}
+			}
 			return false, err
 		}
 		ctr.vecs[i].vec = vec
@@ -101,6 +106,13 @@ func (ctr *container) process(ap *Argument, bat *batch.Batch, proc *process.Proc
 			}
 		}
 	}
+	defer func() {
+		for i := range ctr.vecs {
+			if ctr.vecs[i].needFree {
+				vector.Clean(ctr.vecs[i].vec, proc.Mp())
+			}
+		}
+	}()
 	ovec := ctr.vecs[0].vec
 	var strCol []string
 	n := len(bat.Zs)
diff --git a/pkg/sql/colexec/order/types.go b/pkg/sql/colexec/order/types.go
index f4a335826..1680bb9e6 100644
--- a/pkg/sql/colexec/order/types.go
+++ b/pkg/sql/colexec/order/types.go
@@ -15,10 +15,8 @@
 package order
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 type evalVector struct {
@@ -26,30 +24,13 @@ type evalVector struct {
 	vec      *vector.Vector
 }
 
-type container struct {
+type Container struct {
 	desc      []bool // ds[i] == true: the attrs[i] are in descending order
 	nullsLast []bool
 	vecs      []evalVector // sorted list of attributes
 }
 
 type Argument struct {
-	ctr *container
+	ctr *Container
 	Fs  []*plan.OrderBySpec
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanEvalVectors(mp)
-	}
-}
-
-func (ctr *container) cleanEvalVectors(mp *mpool.MPool) {
-	for i := range ctr.vecs {
-		if ctr.vecs[i].needFree && ctr.vecs[i].vec != nil {
-			ctr.vecs[i].vec.Free(mp)
-			ctr.vecs[i].vec = nil
-		}
-	}
-}
diff --git a/pkg/sql/colexec/output/output.go b/pkg/sql/colexec/output/output.go
index a1f73f593..275b9ff22 100644
--- a/pkg/sql/colexec/output/output.go
+++ b/pkg/sql/colexec/output/output.go
@@ -30,10 +30,7 @@ func Prepare(_ *process.Process, _ any) error {
 
 func Call(_ int, proc *process.Process, arg any) (bool, error) {
 	ap := arg.(*Argument)
-	if bat := proc.Reg.InputBatch; bat != nil {
-		if len(bat.Zs) == 0 {
-			return false, nil
-		}
+	if bat := proc.Reg.InputBatch; bat != nil && len(bat.Zs) > 0 {
 		if err := ap.Func(ap.Data, bat); err != nil {
 			bat.Clean(proc.Mp())
 			return true, err
diff --git a/pkg/sql/colexec/output/types.go b/pkg/sql/colexec/output/types.go
index 0898a11ed..e28355804 100644
--- a/pkg/sql/colexec/output/types.go
+++ b/pkg/sql/colexec/output/types.go
@@ -14,15 +14,9 @@
 
 package output
 
-import (
-	"github.com/matrixorigin/matrixone/pkg/container/batch"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
-)
+import "github.com/matrixorigin/matrixone/pkg/container/batch"
 
 type Argument struct {
 	Data interface{}
 	Func func(interface{}, *batch.Batch) error
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/product/product.go b/pkg/sql/colexec/product/product.go
index 8c43271aa..dd412d5aa 100644
--- a/pkg/sql/colexec/product/product.go
+++ b/pkg/sql/colexec/product/product.go
@@ -50,6 +50,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if len(bat.Zs) == 0 {
diff --git a/pkg/sql/colexec/product/types.go b/pkg/sql/colexec/product/types.go
index 18f1d96d1..7529a0731 100644
--- a/pkg/sql/colexec/product/types.go
+++ b/pkg/sql/colexec/product/types.go
@@ -15,11 +15,9 @@
 package product
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -38,18 +36,3 @@ type Argument struct {
 	Typs   []types.Type
 	Result []colexec.ResultPos
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
diff --git a/pkg/sql/colexec/projection/projection.go b/pkg/sql/colexec/projection/projection.go
index cc3466ca0..9ba2ea414 100644
--- a/pkg/sql/colexec/projection/projection.go
+++ b/pkg/sql/colexec/projection/projection.go
@@ -42,9 +42,8 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 	anal := proc.GetAnalyze(idx)
 	anal.Start()
 	defer anal.Stop()
-	bat := proc.InputBatch()
+	bat := proc.Reg.InputBatch
 	if bat == nil {
-		proc.SetInputBatch(nil)
 		return true, nil
 	}
 	if bat.Length() == 0 {
diff --git a/pkg/sql/colexec/projection/types.go b/pkg/sql/colexec/projection/types.go
index fa5fad349..507b9478b 100644
--- a/pkg/sql/colexec/projection/types.go
+++ b/pkg/sql/colexec/projection/types.go
@@ -14,14 +14,8 @@
 
 package projection
 
-import (
-	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
-)
+import "github.com/matrixorigin/matrixone/pkg/pb/plan"
 
 type Argument struct {
 	Es []*plan.Expr
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/restrict/types.go b/pkg/sql/colexec/restrict/types.go
index b194732c3..c248ecb41 100644
--- a/pkg/sql/colexec/restrict/types.go
+++ b/pkg/sql/colexec/restrict/types.go
@@ -14,14 +14,8 @@
 
 package restrict
 
-import (
-	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
-)
+import "github.com/matrixorigin/matrixone/pkg/pb/plan"
 
 type Argument struct {
 	E *plan.Expr
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
diff --git a/pkg/sql/colexec/semi/join.go b/pkg/sql/colexec/semi/join.go
index 413b2a808..6e08c0bfb 100644
--- a/pkg/sql/colexec/semi/join.go
+++ b/pkg/sql/colexec/semi/join.go
@@ -49,6 +49,7 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 		case Build:
 			if err := ctr.build(ap, proc, anal); err != nil {
 				ctr.state = End
+				ctr.mp.Free()
 				return true, err
 			}
 			ctr.state = Probe
@@ -56,6 +57,10 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				ctr.mp.Free()
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -67,6 +72,7 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			}
 			if err := ctr.probe(bat, ap, proc, anal); err != nil {
 				ctr.state = End
+				ctr.mp.Free()
 				proc.SetInputBatch(nil)
 				return true, err
 			}
@@ -80,10 +86,8 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 
 func (ctr *container) build(ap *Argument, proc *process.Process, anal process.Analyze) error {
 	bat := <-proc.Reg.MergeReceivers[1].Ch
-	if bat != nil {
-		ctr.bat = bat
-		ctr.mp = bat.Ht.(*hashmap.JoinMap).Dup()
-	}
+	ctr.bat = bat
+	ctr.mp = bat.Ht.(*hashmap.JoinMap).Dup()
 	return nil
 }
 
@@ -98,7 +102,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 	if err := ctr.evalJoinCondition(bat, ap.Conditions[0], proc); err != nil {
 		return err
 	}
-	defer ctr.cleanEvalVectors(proc.Mp())
+	defer ctr.freeJoinCondition(proc)
 	count := bat.Length()
 	mSels := ctr.mp.Sels()
 	itr := ctr.mp.Map().NewIterator()
@@ -152,6 +156,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	for i, cond := range conds {
 		vec, err := colexec.EvalExpr(bat, proc, cond)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.Mp())
+				}
+			}
 			return err
 		}
 		ctr.vecs[i] = vec
@@ -166,3 +175,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	}
 	return nil
 }
+
+func (ctr *container) freeJoinCondition(proc *process.Process) {
+	for i := range ctr.evecs {
+		if ctr.evecs[i].needFree {
+			ctr.evecs[i].vec.Free(proc.Mp())
+		}
+	}
+}
diff --git a/pkg/sql/colexec/semi/types.go b/pkg/sql/colexec/semi/types.go
index c9a786c98..3985f1d35 100644
--- a/pkg/sql/colexec/semi/types.go
+++ b/pkg/sql/colexec/semi/types.go
@@ -16,12 +16,10 @@ package semi
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -57,36 +55,3 @@ type Argument struct {
 	Cond       *plan.Expr
 	Conditions [][]*plan.Expr
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-		ctr.cleanEvalVectors(mp)
-		ctr.cleanHashMap()
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.mp != nil {
-		ctr.mp.Free()
-		ctr.mp = nil
-	}
-}
-
-func (ctr *container) cleanEvalVectors(mp *mpool.MPool) {
-	for i := range ctr.evecs {
-		if ctr.evecs[i].needFree && ctr.evecs[i].vec != nil {
-			ctr.evecs[i].vec.Free(mp)
-			ctr.evecs[i].vec = nil
-		}
-	}
-}
diff --git a/pkg/sql/colexec/single/join.go b/pkg/sql/colexec/single/join.go
index 444ba5479..8889001a6 100644
--- a/pkg/sql/colexec/single/join.go
+++ b/pkg/sql/colexec/single/join.go
@@ -55,6 +55,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 		case Build:
 			if err := ctr.build(ap, proc, anal); err != nil {
 				ctr.state = End
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
 				return true, err
 			}
 			ctr.state = Probe
@@ -62,6 +65,12 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			bat := <-proc.Reg.MergeReceivers[0].Ch
 			if bat == nil {
 				ctr.state = End
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.Mp())
+				}
 				continue
 			}
 			if bat.Length() == 0 {
@@ -70,6 +79,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			if ctr.bat.Length() == 0 {
 				if err := ctr.emptyProbe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
 					proc.SetInputBatch(nil)
 					return true, err
 				}
@@ -77,6 +89,9 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 			} else {
 				if err := ctr.probe(bat, ap, proc, anal); err != nil {
 					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
 					proc.SetInputBatch(nil)
 					return true, err
 				}
@@ -130,7 +145,7 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 	if err := ctr.evalJoinCondition(bat, ap.Conditions[0], proc); err != nil {
 		return err
 	}
-	defer ctr.cleanEvalVectors(proc.Mp())
+	defer ctr.freeJoinCondition(proc)
 	count := bat.Length()
 	mSels := ctr.mp.Sels()
 	itr := ctr.mp.Map().NewIterator()
@@ -168,7 +183,6 @@ func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Proces
 					bs := vec.Col.([]bool)
 					if bs[0] {
 						if matched {
-							vec.Free(proc.Mp())
 							return moerr.NewInternalError("scalar subquery returns more than 1 row")
 						}
 						matched = true
@@ -219,6 +233,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	for i, cond := range conds {
 		vec, err := colexec.EvalExpr(bat, proc, cond)
 		if err != nil || vec.ConstExpand(proc.Mp()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.Mp())
+				}
+			}
 			return err
 		}
 		ctr.vecs[i] = vec
@@ -233,3 +252,11 @@ func (ctr *container) evalJoinCondition(bat *batch.Batch, conds []*plan.Expr, pr
 	}
 	return nil
 }
+
+func (ctr *container) freeJoinCondition(proc *process.Process) {
+	for i := range ctr.evecs {
+		if ctr.evecs[i].needFree {
+			ctr.evecs[i].vec.Free(proc.Mp())
+		}
+	}
+}
diff --git a/pkg/sql/colexec/single/types.go b/pkg/sql/colexec/single/types.go
index 7d3ae981d..15d834868 100644
--- a/pkg/sql/colexec/single/types.go
+++ b/pkg/sql/colexec/single/types.go
@@ -16,13 +16,11 @@ package single
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -58,36 +56,3 @@ type Argument struct {
 	Conditions [][]*plan.Expr
 	Result     []colexec.ResultPos
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-		ctr.cleanEvalVectors(mp)
-		ctr.cleanHashMap()
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) cleanHashMap() {
-	if ctr.mp != nil {
-		ctr.mp.Free()
-		ctr.mp = nil
-	}
-}
-
-func (ctr *container) cleanEvalVectors(mp *mpool.MPool) {
-	for i := range ctr.evecs {
-		if ctr.evecs[i].needFree && ctr.evecs[i].vec != nil {
-			ctr.evecs[i].vec.Free(mp)
-			ctr.evecs[i].vec = nil
-		}
-	}
-}
diff --git a/pkg/sql/colexec/top/top.go b/pkg/sql/colexec/top/top.go
index 277d2371d..8e84927ff 100644
--- a/pkg/sql/colexec/top/top.go
+++ b/pkg/sql/colexec/top/top.go
@@ -41,7 +41,7 @@ func String(arg any, buf *bytes.Buffer) {
 
 func Prepare(_ *process.Process, arg any) error {
 	ap := arg.(*Argument)
-	ap.ctr = new(container)
+	ap.ctr = new(Container)
 	if ap.Limit > 1024 {
 		ap.ctr.sels = make([]int64, 0, 1024)
 	} else {
@@ -88,7 +88,7 @@ func Call(idx int, proc *process.Process, arg any) (bool, error) {
 	}
 }
 
-func (ctr *container) build(ap *Argument, bat *batch.Batch, proc *process.Process) error {
+func (ctr *Container) build(ap *Argument, bat *batch.Batch, proc *process.Process) error {
 	ctr.n = len(bat.Vecs)
 	ctr.poses = ctr.poses[:0]
 	for _, f := range ap.Fs {
@@ -139,7 +139,7 @@ func (ctr *container) build(ap *Argument, bat *batch.Batch, proc *process.Proces
 	return ctr.processBatch(ap.Limit, bat, proc)
 }
 
-func (ctr *container) processBatch(limit int64, bat *batch.Batch, proc *process.Process) error {
+func (ctr *Container) processBatch(limit int64, bat *batch.Batch, proc *process.Process) error {
 	var start int64
 
 	length := int64(len(bat.Zs))
@@ -151,6 +151,7 @@ func (ctr *container) processBatch(limit int64, bat *batch.Batch, proc *process.
 		for i := int64(0); i < start; i++ {
 			for j, vec := range ctr.bat.Vecs {
 				if err := vector.UnionOne(vec, bat.Vecs[j], i, proc.Mp()); err != nil {
+					ctr.bat.Clean(proc.Mp())
 					return err
 				}
 			}
@@ -174,6 +175,7 @@ func (ctr *container) processBatch(limit int64, bat *batch.Batch, proc *process.
 		if ctr.compare(1, 0, i, ctr.sels[0]) < 0 {
 			for _, cmp := range ctr.cmps {
 				if err := cmp.Copy(1, 0, i, ctr.sels[0], proc); err != nil {
+					ctr.bat.Clean(proc.Mp())
 					return err
 				}
 				ctr.bat.Zs[0] = bat.Zs[i]
@@ -184,7 +186,7 @@ func (ctr *container) processBatch(limit int64, bat *batch.Batch, proc *process.
 	return nil
 }
 
-func (ctr *container) eval(limit int64, proc *process.Process) error {
+func (ctr *Container) eval(limit int64, proc *process.Process) error {
 	if int64(len(ctr.sels)) < limit {
 		ctr.sort()
 	}
@@ -196,10 +198,11 @@ func (ctr *container) eval(limit int64, proc *process.Process) error {
 		sels[len(sels)-1-i] = heap.Pop(ctr).(int64)
 	}
 	if err := ctr.bat.Shuffle(sels, proc.Mp()); err != nil {
-		return err
+		ctr.bat.Clean(proc.Mp())
+		ctr.bat = nil
 	}
 	for i := ctr.n; i < len(ctr.bat.Vecs); i++ {
-		ctr.bat.Vecs[i].Free(proc.Mp())
+		vector.Clean(ctr.bat.Vecs[i], proc.Mp())
 	}
 	ctr.bat.Vecs = ctr.bat.Vecs[:ctr.n]
 	ctr.bat.ExpandNulls()
@@ -209,7 +212,7 @@ func (ctr *container) eval(limit int64, proc *process.Process) error {
 }
 
 // do sort work for heap, and result order will be set in container.sels
-func (ctr *container) sort() {
+func (ctr *Container) sort() {
 	for i, cmp := range ctr.cmps {
 		cmp.Set(0, ctr.bat.Vecs[i])
 	}
diff --git a/pkg/sql/colexec/top/types.go b/pkg/sql/colexec/top/types.go
index 2aae2d1bd..37e0e58b0 100644
--- a/pkg/sql/colexec/top/types.go
+++ b/pkg/sql/colexec/top/types.go
@@ -15,11 +15,9 @@
 package top
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/compare"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 const (
@@ -28,7 +26,7 @@ const (
 	End
 )
 
-type container struct {
+type Container struct {
 	n     int // result vector number
 	state int
 	sels  []int64
@@ -40,26 +38,11 @@ type container struct {
 
 type Argument struct {
 	Limit int64
-	ctr   *container
+	ctr   *Container
 	Fs    []*plan.OrderBySpec
 }
 
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-	ctr := arg.ctr
-	if ctr != nil {
-		mp := proc.Mp()
-		ctr.cleanBatch(mp)
-	}
-}
-
-func (ctr *container) cleanBatch(mp *mpool.MPool) {
-	if ctr.bat != nil {
-		ctr.bat.Clean(mp)
-		ctr.bat = nil
-	}
-}
-
-func (ctr *container) compare(vi, vj int, i, j int64) int {
+func (ctr *Container) compare(vi, vj int, i, j int64) int {
 	for _, pos := range ctr.poses {
 		if r := ctr.cmps[pos].Compare(vi, vj, i, j); r != 0 {
 			return r
@@ -68,23 +51,23 @@ func (ctr *container) compare(vi, vj int, i, j int64) int {
 	return 0
 }
 
-func (ctr *container) Len() int {
+func (ctr *Container) Len() int {
 	return len(ctr.sels)
 }
 
-func (ctr *container) Less(i, j int) bool {
+func (ctr *Container) Less(i, j int) bool {
 	return ctr.compare(0, 0, ctr.sels[i], ctr.sels[j]) > 0
 }
 
-func (ctr *container) Swap(i, j int) {
+func (ctr *Container) Swap(i, j int) {
 	ctr.sels[i], ctr.sels[j] = ctr.sels[j], ctr.sels[i]
 }
 
-func (ctr *container) Push(x interface{}) {
+func (ctr *Container) Push(x interface{}) {
 	ctr.sels = append(ctr.sels, x.(int64))
 }
 
-func (ctr *container) Pop() interface{} {
+func (ctr *Container) Pop() interface{} {
 	n := len(ctr.sels) - 1
 	x := ctr.sels[n]
 	ctr.sels = ctr.sels[:n]
diff --git a/pkg/sql/colexec/unnest/types.go b/pkg/sql/colexec/unnest/types.go
index 8e8596beb..965e3d0f1 100644
--- a/pkg/sql/colexec/unnest/types.go
+++ b/pkg/sql/colexec/unnest/types.go
@@ -16,7 +16,6 @@ package unnest
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 type Param struct {
@@ -34,10 +33,6 @@ type Param struct {
 type Argument struct {
 	Es *Param
 }
-
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
-
 type ExternalParam struct {
 	ColName string
 	Path    string
diff --git a/pkg/sql/colexec/unnest/unnest.go b/pkg/sql/colexec/unnest/unnest.go
index 3d7066e03..9680e7f1b 100644
--- a/pkg/sql/colexec/unnest/unnest.go
+++ b/pkg/sql/colexec/unnest/unnest.go
@@ -72,10 +72,8 @@ func Call(_ int, proc *process.Process, arg any) (bool, error) {
 func callByStr(param *Param, proc *process.Process) (bool, error) {
 	bat := proc.InputBatch()
 	if bat == nil {
-		proc.SetInputBatch(nil)
 		return true, nil
 	}
-	defer bat.Clean(proc.Mp())
 	json, err := types.ParseStringToByteJson(bat.Vecs[0].GetString(0))
 	if err != nil {
 		return false, err
@@ -89,13 +87,11 @@ func callByStr(param *Param, proc *process.Process) (bool, error) {
 		return false, err
 	}
 	rbat := batch.New(false, param.Attrs)
-	rbat.Cnt = 1
 	for i := range param.Cols {
 		rbat.Vecs[i] = vector.New(dupType(param.Cols[i].Typ))
 	}
 	rbat, err = makeBatch(rbat, ures, param, proc)
 	if err != nil {
-		rbat.Clean(proc.Mp())
 		return false, err
 	}
 	rbat.InitZsOne(len(ures))
@@ -106,10 +102,8 @@ func callByStr(param *Param, proc *process.Process) (bool, error) {
 func callByCol(param *Param, proc *process.Process) (bool, error) {
 	bat := proc.InputBatch()
 	if bat == nil {
-		proc.SetInputBatch(nil)
 		return true, nil
 	}
-	defer bat.Clean(proc.Mp())
 	if len(bat.Vecs) != 1 {
 		return false, moerr.NewInvalidArg("unnest: invalid input batch,len(vecs)[%d] != 1", len(bat.Vecs))
 	}
@@ -122,7 +116,6 @@ func callByCol(param *Param, proc *process.Process) (bool, error) {
 		return false, err
 	}
 	rbat := batch.New(false, param.Attrs)
-	rbat.Cnt = 1
 	for i := range param.Cols {
 		rbat.Vecs[i] = vector.New(dupType(param.Cols[i].Typ))
 	}
@@ -136,7 +129,6 @@ func callByCol(param *Param, proc *process.Process) (bool, error) {
 		}
 		rbat, err = makeBatch(rbat, ures, param, proc)
 		if err != nil {
-			rbat.Clean(proc.Mp())
 			return false, err
 		}
 		rows += len(ures)
@@ -179,9 +171,10 @@ func makeBatch(bat *batch.Batch, ures []bytejson.UnnestResult, param *Param, pro
 	return bat, nil
 }
 func dupType(typ *plan.Type) types.Type {
-	return types.New(
-		types.T(typ.Id),
-		typ.Width,
-		typ.Size,
-		typ.Precision)
+	return types.Type{
+		Oid:       types.T(typ.Id),
+		Width:     typ.Width,
+		Size:      typ.Size,
+		Precision: typ.Precision,
+	}
 }
diff --git a/pkg/sql/colexec/update/types.go b/pkg/sql/colexec/update/types.go
index 7d6e50ef3..fd0b2e929 100644
--- a/pkg/sql/colexec/update/types.go
+++ b/pkg/sql/colexec/update/types.go
@@ -17,7 +17,6 @@ package update
 import (
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 type Argument struct {
@@ -30,9 +29,6 @@ type Argument struct {
 	TableID      []string
 }
 
-func (arg *Argument) Free(proc *process.Process, pipelineFailed bool) {
-}
-
 type UpdateCtx struct {
 	PriKey             string
 	PriKeyIdx          int32 // delete if -1
diff --git a/pkg/sql/compile/compile.go b/pkg/sql/compile/compile.go
index b5d37136a..3b9c9bd9e 100644
--- a/pkg/sql/compile/compile.go
+++ b/pkg/sql/compile/compile.go
@@ -688,7 +688,7 @@ func (c *Compile) compileJoin(n, right *plan.Node, ss []*Scope, children []*Scop
 	isEq := isEquiJoin(n.OnList)
 	typs := make([]types.Type, len(right.ProjectList))
 	for i, expr := range right.ProjectList {
-		typs[i] = plan2.MakeTypeByPlan2Type(expr.Typ)
+		typs[i] = dupType(expr.Typ)
 	}
 	switch joinTyp {
 	case plan.Node_INNER:
@@ -811,9 +811,6 @@ func (c *Compile) compileSort(n *plan.Node, ss []*Scope) []*Scope {
 
 func (c *Compile) compileTop(n *plan.Node, ss []*Scope) []*Scope {
 	for i := range ss {
-		if containBrokenNode(ss[i]) {
-			ss[i] = c.newMergeScope([]*Scope{ss[i]})
-		}
 		ss[i].appendInstruction(vm.Instruction{
 			Op:  vm.Top,
 			Idx: c.anal.curr,
@@ -831,9 +828,6 @@ func (c *Compile) compileTop(n *plan.Node, ss []*Scope) []*Scope {
 
 func (c *Compile) compileOrder(n *plan.Node, ss []*Scope) []*Scope {
 	for i := range ss {
-		if containBrokenNode(ss[i]) {
-			ss[i] = c.newMergeScope([]*Scope{ss[i]})
-		}
 		ss[i].appendInstruction(vm.Instruction{
 			Op:  vm.Order,
 			Idx: c.anal.curr,
@@ -850,11 +844,6 @@ func (c *Compile) compileOrder(n *plan.Node, ss []*Scope) []*Scope {
 }
 
 func (c *Compile) compileOffset(n *plan.Node, ss []*Scope) []*Scope {
-	for i := range ss {
-		if containBrokenNode(ss[i]) {
-			ss[i] = c.newMergeScope([]*Scope{ss[i]})
-		}
-	}
 	rs := c.newMergeScope(ss)
 	rs.Instructions[0] = vm.Instruction{
 		Op:  vm.MergeOffset,
@@ -866,9 +855,6 @@ func (c *Compile) compileOffset(n *plan.Node, ss []*Scope) []*Scope {
 
 func (c *Compile) compileLimit(n *plan.Node, ss []*Scope) []*Scope {
 	for i := range ss {
-		if containBrokenNode(ss[i]) {
-			ss[i] = c.newMergeScope([]*Scope{ss[i]})
-		}
 		ss[i].appendInstruction(vm.Instruction{
 			Op:  vm.Limit,
 			Idx: c.anal.curr,
@@ -884,20 +870,8 @@ func (c *Compile) compileLimit(n *plan.Node, ss []*Scope) []*Scope {
 	return []*Scope{rs}
 }
 
-func containBrokenNode(s *Scope) bool {
-	for i := range s.Instructions {
-		if s.Instructions[i].IsBrokenNode() {
-			return true
-		}
-	}
-	return false
-}
-
 func (c *Compile) compileAgg(n *plan.Node, ss []*Scope, ns []*plan.Node) []*Scope {
 	for i := range ss {
-		if containBrokenNode(ss[i]) {
-			ss[i] = c.newMergeScope([]*Scope{ss[i]})
-		}
 		ss[i].appendInstruction(vm.Instruction{
 			Op:  vm.Group,
 			Idx: c.anal.curr,
@@ -917,11 +891,6 @@ func (c *Compile) compileGroup(n *plan.Node, ss []*Scope, ns []*plan.Node) []*Sc
 	rs := c.newScopeList(validScopeCount(ss))
 	j := 0
 	for i := range ss {
-		if containBrokenNode(ss[i]) {
-			isEnd := ss[i].IsEnd
-			ss[i] = c.newMergeScope([]*Scope{ss[i]})
-			ss[i].IsEnd = isEnd
-		}
 		if !ss[i].IsEnd {
 			ss[i].appendInstruction(vm.Instruction{
 				Op:  vm.Dispatch,
@@ -1253,3 +1222,13 @@ func joinType(n *plan.Node, ns []*plan.Node) (bool, plan.Node_JoinFlag) {
 		panic(moerr.NewNYI(fmt.Sprintf("join typ '%v'", n.JoinType)))
 	}
 }
+
+func dupType(typ *plan.Type) types.Type {
+	return types.Type{
+		Oid:       types.T(typ.Id),
+		Size:      typ.Size,
+		Width:     typ.Width,
+		Scale:     typ.Scale,
+		Precision: typ.Precision,
+	}
+}
diff --git a/pkg/sql/plan/make.go b/pkg/sql/plan/make.go
index 1f997303d..37342a38c 100644
--- a/pkg/sql/plan/make.go
+++ b/pkg/sql/plan/make.go
@@ -222,8 +222,6 @@ func makePlan2Type(typ *types.Type) *plan.Type {
 	}
 }
 
-var MakeTypeByPlan2Type = makeTypeByPlan2Type
-
 func makeTypeByPlan2Type(typ *plan.Type) types.Type {
 	return types.Type{
 		Oid:       types.T(typ.Id),
diff --git a/pkg/vm/pipeline/pipeline.go b/pkg/vm/pipeline/pipeline.go
index 29c4aec06..b110a2f9a 100644
--- a/pkg/vm/pipeline/pipeline.go
+++ b/pkg/vm/pipeline/pipeline.go
@@ -17,7 +17,10 @@ package pipeline
 import (
 	"bytes"
 
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/dispatch"
+
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/connector"
 	"github.com/matrixorigin/matrixone/pkg/vm"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
@@ -50,6 +53,7 @@ func (p *Pipeline) Run(r engine.Reader, proc *process.Process) (bool, error) {
 	var err error
 	var bat *batch.Batch
 
+	defer cleanup(p, proc)
 	if p.reg != nil { // used to handle some push-down request
 		select {
 		case <-p.reg.Ctx.Done():
@@ -57,13 +61,11 @@ func (p *Pipeline) Run(r engine.Reader, proc *process.Process) (bool, error) {
 		}
 	}
 	if err = vm.Prepare(p.instructions, proc); err != nil {
-		cleanup(p, proc, false)
 		return false, err
 	}
 	for {
 		// read data from storage engine
 		if bat, err = r.Read(p.attrs, nil, proc.Mp()); err != nil {
-			cleanup(p, proc, false)
 			return false, err
 		}
 		if bat != nil {
@@ -71,15 +73,9 @@ func (p *Pipeline) Run(r engine.Reader, proc *process.Process) (bool, error) {
 		}
 		// processing the batch according to the instructions
 		proc.Reg.InputBatch = bat
-		end, err = vm.Run(p.instructions, proc)
-		if err != nil {
-			cleanup(p, proc, false)
+		if end, err = vm.Run(p.instructions, proc); err != nil || end { // end is true means pipeline successfully completed
 			return end, err
 		}
-		if end {
-			cleanup(p, proc, true)
-			return end, nil
-		}
 	}
 }
 
@@ -87,6 +83,7 @@ func (p *Pipeline) ConstRun(bat *batch.Batch, proc *process.Process) (bool, erro
 	var end bool // exist flag
 	var err error
 
+	defer cleanup(p, proc)
 	if p.reg != nil { // used to handle some push-down request
 		select {
 		case <-p.reg.Ctx.Done():
@@ -98,19 +95,14 @@ func (p *Pipeline) ConstRun(bat *batch.Batch, proc *process.Process) (bool, erro
 	}
 	bat.Cnt = 1
 	// processing the batch according to the instructions
-	pipelineInputs := []*batch.Batch{bat, nil}
 	for {
-		for i := range pipelineInputs {
-			proc.Reg.InputBatch = pipelineInputs[i]
-			end, err = vm.Run(p.instructions, proc)
-			if err != nil {
-				cleanup(p, proc, false)
-				return end, err
-			}
-			if end {
-				cleanup(p, proc, true)
-				return end, nil
-			}
+		proc.Reg.InputBatch = bat
+		if end, err = vm.Run(p.instructions, proc); err != nil || end {
+			return end, err
+		}
+		proc.Reg.InputBatch = nil
+		if end, err = vm.Run(p.instructions, proc); err != nil || end {
+			return end, err
 		}
 	}
 }
@@ -119,7 +111,28 @@ func (p *Pipeline) MergeRun(proc *process.Process) (bool, error) {
 	var end bool
 	var err error
 
-	defer proc.Cancel()
+	// XXX Here is the big problem.   In side defer, we call cleanup
+	// which in turn can calls Run.   Using defer to trigger normal
+	// execution flow is simply WRONG.   Calling Run in cleanup, at
+	// best is extremely bad naming.
+	//
+	// I have observed a panic within, calls defer, calls cleanup,
+	// calls Run, may create a deadlock.
+	//
+	// Will try to repro it with fault inj.
+	//
+	defer func() {
+		cleanup(p, proc)
+		for i := 0; i < len(proc.Reg.MergeReceivers); i++ { // simulating the end of a pipeline
+			for len(proc.Reg.MergeReceivers[i].Ch) > 0 {
+				bat := <-proc.Reg.MergeReceivers[i].Ch
+				if bat != nil {
+					bat.Clean(proc.Mp())
+				}
+			}
+		}
+		proc.Cancel()
+	}()
 	if p.reg != nil { // used to handle some push-down request
 		select {
 		case <-p.reg.Ctx.Done():
@@ -127,46 +140,43 @@ func (p *Pipeline) MergeRun(proc *process.Process) (bool, error) {
 		}
 	}
 	if err := vm.Prepare(p.instructions, proc); err != nil {
-		cleanup(p, proc, false)
 		return false, err
 	}
 	for {
 		proc.Reg.InputBatch = nil
-		end, err = vm.Run(p.instructions, proc)
-		if err != nil {
-			cleanup(p, proc, false)
+		if end, err = vm.Run(p.instructions, proc); err != nil || end {
 			return end, err
 		}
-		if end {
-			cleanup(p, proc, true)
-			return end, nil
-		}
 	}
 }
 
-// cleanup do memory release work for a whole pipeline.
-// clean the coming batches and template space of each pipeline operator.
-func cleanup(p *Pipeline, proc *process.Process, pipelineSucceed bool) {
-	// clean all the coming batches.
-	if !pipelineSucceed {
-		bat := proc.InputBatch()
-		if bat != nil {
-			bat.Clean(proc.Mp())
-		}
-		proc.SetInputBatch(nil)
-	}
-	for i := range proc.Reg.MergeReceivers {
-		for len(proc.Reg.MergeReceivers[i].Ch) > 0 {
-			bat := <-proc.Reg.MergeReceivers[i].Ch
-			if bat == nil {
+func cleanup(p *Pipeline, proc *process.Process) {
+	proc.Reg.InputBatch = nil
+	_, _ = vm.Run(p.instructions, proc)
+	for i, in := range p.instructions {
+		if in.Op == vm.Connector {
+			arg := p.instructions[i].Arg.(*connector.Argument)
+			if len(arg.Reg.Ch) > 0 {
 				break
 			}
-			bat.Clean(proc.Mp())
+			select {
+			case <-arg.Reg.Ctx.Done():
+			case arg.Reg.Ch <- nil:
+			}
+			break
+		}
+		if in.Op == vm.Dispatch {
+			arg := p.instructions[i].Arg.(*dispatch.Argument)
+			for _, reg := range arg.Regs {
+				if len(reg.Ch) > 0 {
+					break
+				}
+				select {
+				case <-reg.Ctx.Done():
+				case reg.Ch <- nil:
+				}
+			}
+			break
 		}
-	}
-
-	// clean operator space.
-	for i := range p.instructions {
-		p.instructions[i].Arg.Free(proc, !pipelineSucceed)
 	}
 }
diff --git a/pkg/vm/types.go b/pkg/vm/types.go
index fa3645e15..14a414c50 100644
--- a/pkg/vm/types.go
+++ b/pkg/vm/types.go
@@ -14,8 +14,6 @@
 
 package vm
 
-import "github.com/matrixorigin/matrixone/pkg/vm/process"
-
 const (
 	Top = iota
 	Join
@@ -64,11 +62,6 @@ const (
 	GenerateSeries
 )
 
-type InstructionArgument interface {
-	// Free release all memory allocated from mPool in an operator.
-	Free(proc *process.Process, pipelineFailed bool)
-}
-
 // Instruction contains relational algebra
 type Instruction struct {
 	// Op specified the operator code of an instruction.
@@ -76,23 +69,7 @@ type Instruction struct {
 	// Idx specified the anaylze information index.
 	Idx int
 	// Arg contains the operand of this instruction.
-	Arg InstructionArgument
+	Arg any
 }
 
 type Instructions []Instruction
-
-func (ins *Instruction) IsBrokenNode() bool {
-	switch ins.Op {
-	case Order, MergeOrder:
-		return true
-	case Limit, MergeLimit:
-		return true
-	case Offset, MergeOffset:
-		return true
-	case Group, MergeGroup:
-		return true
-	case Top, MergeTop:
-		return true
-	}
-	return false
-}


Test patch:
diff --git a/pkg/sql/colexec/anti/join_test.go b/pkg/sql/colexec/anti/join_test.go
index 93cb3efd8..d25a8e1dd 100644
--- a/pkg/sql/colexec/anti/join_test.go
+++ b/pkg/sql/colexec/anti/join_test.go
@@ -97,7 +97,6 @@ func TestAnti(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 	for _, tc := range tcs {
@@ -116,7 +115,6 @@ func TestAnti(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/connector/connector_test.go b/pkg/sql/colexec/connector/connector_test.go
index f4c1842df..f3b3939ad 100644
--- a/pkg/sql/colexec/connector/connector_test.go
+++ b/pkg/sql/colexec/connector/connector_test.go
@@ -81,7 +81,7 @@ func TestConnector(t *testing.T) {
 		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = nil
 		_, _ = Call(0, tc.proc, tc.arg)
-		for len(tc.arg.Reg.Ch) > 0 {
+		for {
 			bat := <-tc.arg.Reg.Ch
 			if bat == nil {
 				break
@@ -91,7 +91,6 @@ func TestConnector(t *testing.T) {
 			}
 			bat.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/dispatch/dispatch_test.go b/pkg/sql/colexec/dispatch/dispatch_test.go
index b06e2c747..232fcdd32 100644
--- a/pkg/sql/colexec/dispatch/dispatch_test.go
+++ b/pkg/sql/colexec/dispatch/dispatch_test.go
@@ -82,7 +82,7 @@ func TestDispatch(t *testing.T) {
 		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = nil
 		_, _ = Call(0, tc.proc, tc.arg)
-		for len(tc.arg.Regs[0].Ch) > 0 {
+		for {
 			bat := <-tc.arg.Regs[0].Ch
 			if bat == nil {
 				break
diff --git a/pkg/sql/colexec/group/group_test.go b/pkg/sql/colexec/group/group_test.go
index 893494cf0..6243a49fc 100644
--- a/pkg/sql/colexec/group/group_test.go
+++ b/pkg/sql/colexec/group/group_test.go
@@ -116,7 +116,6 @@ func TestGroup(t *testing.T) {
 		tc.proc.Reg.InputBatch = nil
 		_, err = Call(0, tc.proc, tc.arg)
 		require.NoError(t, err)
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/hashbuild/build_test.go b/pkg/sql/colexec/hashbuild/build_test.go
index fbd7f72e0..7fafe1356 100644
--- a/pkg/sql/colexec/hashbuild/build_test.go
+++ b/pkg/sql/colexec/hashbuild/build_test.go
@@ -83,7 +83,6 @@ func TestBuild(t *testing.T) {
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 			break
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/intersect/intersect_test.go b/pkg/sql/colexec/intersect/intersect_test.go
index b3472868b..b5b9cf897 100644
--- a/pkg/sql/colexec/intersect/intersect_test.go
+++ b/pkg/sql/colexec/intersect/intersect_test.go
@@ -91,7 +91,6 @@ func TestIntersect(t *testing.T) {
 		}
 	}
 	require.Equal(t, 1, cnt) // 1 row
-	c.arg.Free(c.proc, false)
 	require.Equal(t, int64(0), c.proc.Mp().CurrNB())
 }
 
diff --git a/pkg/sql/colexec/intersectall/intersectall_test.go b/pkg/sql/colexec/intersectall/intersectall_test.go
index 0383da474..0644320aa 100644
--- a/pkg/sql/colexec/intersectall/intersectall_test.go
+++ b/pkg/sql/colexec/intersectall/intersectall_test.go
@@ -87,10 +87,11 @@ func TestIntersectAll(t *testing.T) {
 			cnt += result.Length()
 			require.Equal(t, 3, len(result.Vecs))
 			c.proc.InputBatch().Clean(c.proc.Mp())
-		}
+		} /*else {
+			c.proc.InputBatch().Clean(c.proc.Mp())
+		}*/
 	}
 	require.Equal(t, 2, cnt) // 1 row
-	c.arg.Free(c.proc, false)
 	require.Equal(t, int64(0), c.proc.Mp().CurrNB())
 }
 
diff --git a/pkg/sql/colexec/join/join_test.go b/pkg/sql/colexec/join/join_test.go
index 514940b0b..9dbe008a8 100644
--- a/pkg/sql/colexec/join/join_test.go
+++ b/pkg/sql/colexec/join/join_test.go
@@ -99,7 +99,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		nb1 := tc.proc.Mp().CurrNB()
 		require.Equal(t, nb0, nb1)
 	}
diff --git a/pkg/sql/colexec/left/join_test.go b/pkg/sql/colexec/left/join_test.go
index 414fbaa72..1c6fc05ad 100644
--- a/pkg/sql/colexec/left/join_test.go
+++ b/pkg/sql/colexec/left/join_test.go
@@ -99,7 +99,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		nb1 := tc.proc.Mp().CurrNB()
 		require.Equal(t, nb0, nb1)
 	}
@@ -120,7 +119,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		nb1 := tc.proc.Mp().CurrNB()
 		require.Equal(t, nb0, nb1)
 	}
diff --git a/pkg/sql/colexec/limit/limit_test.go b/pkg/sql/colexec/limit/limit_test.go
index b756e619f..465402cee 100644
--- a/pkg/sql/colexec/limit/limit_test.go
+++ b/pkg/sql/colexec/limit/limit_test.go
@@ -109,7 +109,6 @@ func TestLimit(t *testing.T) {
 		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = nil
 		_, _ = Call(0, tc.proc, tc.arg)
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/loopanti/join_test.go b/pkg/sql/colexec/loopanti/join_test.go
index 024a9ed6d..049e2d26e 100644
--- a/pkg/sql/colexec/loopanti/join_test.go
+++ b/pkg/sql/colexec/loopanti/join_test.go
@@ -88,7 +88,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 	for _, tc := range tcs {
@@ -107,7 +106,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 
diff --git a/pkg/sql/colexec/loopjoin/join_test.go b/pkg/sql/colexec/loopjoin/join_test.go
index cba25c6a9..6e1a4550b 100644
--- a/pkg/sql/colexec/loopjoin/join_test.go
+++ b/pkg/sql/colexec/loopjoin/join_test.go
@@ -88,7 +88,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/loopleft/join_test.go b/pkg/sql/colexec/loopleft/join_test.go
index cf834df12..3d7e0c31a 100644
--- a/pkg/sql/colexec/loopleft/join_test.go
+++ b/pkg/sql/colexec/loopleft/join_test.go
@@ -89,7 +89,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 	for _, tc := range tcs {
@@ -108,7 +107,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 
diff --git a/pkg/sql/colexec/loopsemi/join_test.go b/pkg/sql/colexec/loopsemi/join_test.go
index daa2dcba5..6dfdcc2f8 100644
--- a/pkg/sql/colexec/loopsemi/join_test.go
+++ b/pkg/sql/colexec/loopsemi/join_test.go
@@ -88,7 +88,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/loopsingle/join_test.go b/pkg/sql/colexec/loopsingle/join_test.go
index bb00184f5..fc7afba87 100644
--- a/pkg/sql/colexec/loopsingle/join_test.go
+++ b/pkg/sql/colexec/loopsingle/join_test.go
@@ -89,7 +89,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 	for _, tc := range tcs {
@@ -108,7 +107,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 
diff --git a/pkg/sql/colexec/mark/join_test.go b/pkg/sql/colexec/mark/join_test.go
index ee779bf3f..439f98610 100644
--- a/pkg/sql/colexec/mark/join_test.go
+++ b/pkg/sql/colexec/mark/join_test.go
@@ -82,27 +82,27 @@ func TestString(t *testing.T) {
 
 func TestMark(t *testing.T) {
 	/* XXX There are some problems with the mark join code for handling null. Modify later
-	   for _, tc := range tcs {
-	   	bat := hashBuild(t, tc)
-	   	err := Prepare(tc.proc, tc.arg)
-	   	require.NoError(t, err)
-	   	batWithNull := newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
-	   	batWithNull.Vecs[0].Nsp.Np.Add(0)
-	   	tc.proc.Reg.MergeReceivers[0].Ch <- batWithNull
-	   	tc.proc.Reg.MergeReceivers[0].Ch <- &batch.Batch{}
-	   	tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
-	   	tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
-	   	tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
-	   	tc.proc.Reg.MergeReceivers[0].Ch <- nil
-	   	tc.proc.Reg.MergeReceivers[1].Ch <- bat
-	   	for {
-	   		if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
-	   			break
-	   		}
-	   		tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
-	   	}
-	   	require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
-	   }
+	for _, tc := range tcs {
+		bat := hashBuild(t, tc)
+		err := Prepare(tc.proc, tc.arg)
+		require.NoError(t, err)
+		batWithNull := newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		batWithNull.Vecs[0].Nsp.Np.Add(0)
+		tc.proc.Reg.MergeReceivers[0].Ch <- batWithNull
+		tc.proc.Reg.MergeReceivers[0].Ch <- &batch.Batch{}
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- nil
+		tc.proc.Reg.MergeReceivers[1].Ch <- bat
+		for {
+			if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
+				break
+			}
+			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
+		}
+		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
+	}
 	*/
 	for _, tc := range tcs {
 		err := Prepare(tc.proc, tc.arg)
@@ -120,7 +120,6 @@ func TestMark(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/merge/merge_test.go b/pkg/sql/colexec/merge/merge_test.go
index 0b9549a80..c8f0a9445 100644
--- a/pkg/sql/colexec/merge/merge_test.go
+++ b/pkg/sql/colexec/merge/merge_test.go
@@ -92,7 +92,6 @@ func TestMerge(t *testing.T) {
 				}
 			}
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/mergegroup/group_test.go b/pkg/sql/colexec/mergegroup/group_test.go
index 034b0c4ea..4ef4b1ea6 100644
--- a/pkg/sql/colexec/mergegroup/group_test.go
+++ b/pkg/sql/colexec/mergegroup/group_test.go
@@ -100,7 +100,6 @@ func TestGroup(t *testing.T) {
 	for _, tc := range tcs {
 		err := Prepare(tc.proc, tc.arg)
 		require.NoError(t, err)
-		oldMpool := tc.proc.Mp().CurrNB()
 		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
 		tc.proc.Reg.MergeReceivers[0].Ch <- &batch.Batch{}
 		tc.proc.Reg.MergeReceivers[0].Ch <- nil
@@ -114,9 +113,6 @@ func TestGroup(t *testing.T) {
 				}
 				break
 			}
-			if bat := tc.proc.InputBatch(); bat != nil {
-				bat.Clean(tc.proc.Mp())
-			}
 		}
 		for i := 0; i < len(tc.proc.Reg.MergeReceivers); i++ { // simulating the end of a pipeline
 			for len(tc.proc.Reg.MergeReceivers[i].Ch) > 0 {
@@ -126,8 +122,7 @@ func TestGroup(t *testing.T) {
 				}
 			}
 		}
-		tc.arg.Free(tc.proc, true)
-		require.Equal(t, oldMpool, tc.proc.Mp().CurrNB())
+		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
 
diff --git a/pkg/sql/colexec/mergelimit/limit_test.go b/pkg/sql/colexec/mergelimit/limit_test.go
index 5031cdace..540d0f0e8 100644
--- a/pkg/sql/colexec/mergelimit/limit_test.go
+++ b/pkg/sql/colexec/mergelimit/limit_test.go
@@ -95,7 +95,6 @@ func TestLimit(t *testing.T) {
 				}
 			}
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/mergeoffset/offset_test.go b/pkg/sql/colexec/mergeoffset/offset_test.go
index 6b51940e0..d5c30bd2b 100644
--- a/pkg/sql/colexec/mergeoffset/offset_test.go
+++ b/pkg/sql/colexec/mergeoffset/offset_test.go
@@ -95,7 +95,6 @@ func TestOffset(t *testing.T) {
 				}
 			}
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/mergeorder/order_test.go b/pkg/sql/colexec/mergeorder/order_test.go
index eb5b9ffbb..e976ceec2 100644
--- a/pkg/sql/colexec/mergeorder/order_test.go
+++ b/pkg/sql/colexec/mergeorder/order_test.go
@@ -87,9 +87,6 @@ func TestOrder(t *testing.T) {
 				}
 				break
 			}
-			if bat := tc.proc.InputBatch(); bat != nil {
-				bat.Clean(tc.proc.Mp())
-			}
 		}
 		for i := 0; i < len(tc.proc.Reg.MergeReceivers); i++ { // simulating the end of a pipeline
 			for len(tc.proc.Reg.MergeReceivers[i].Ch) > 0 {
@@ -99,7 +96,6 @@ func TestOrder(t *testing.T) {
 				}
 			}
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, tc.proc.Mp().CurrNB(), int64(0))
 	}
 }
diff --git a/pkg/sql/colexec/mergetop/top_test.go b/pkg/sql/colexec/mergetop/top_test.go
index c3b5a13ab..0587411c7 100644
--- a/pkg/sql/colexec/mergetop/top_test.go
+++ b/pkg/sql/colexec/mergetop/top_test.go
@@ -88,9 +88,6 @@ func TestTop(t *testing.T) {
 				}
 				break
 			}
-			if bat := tc.proc.InputBatch(); bat != nil {
-				bat.Clean(tc.proc.Mp())
-			}
 		}
 		for i := 0; i < len(tc.proc.Reg.MergeReceivers); i++ { // simulating the end of a pipeline
 			for len(tc.proc.Reg.MergeReceivers[i].Ch) > 0 {
@@ -100,7 +97,6 @@ func TestTop(t *testing.T) {
 				}
 			}
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, tc.proc.Mp().CurrNB(), int64(0))
 	}
 }
diff --git a/pkg/sql/colexec/minus/minus_test.go b/pkg/sql/colexec/minus/minus_test.go
index 65bd5e3de..b561779c5 100644
--- a/pkg/sql/colexec/minus/minus_test.go
+++ b/pkg/sql/colexec/minus/minus_test.go
@@ -91,7 +91,6 @@ func TestMinus(t *testing.T) {
 		}
 	}
 	require.Equal(t, 1, cnt) // 1 row
-	c.arg.Free(c.proc, false)
 	require.Equal(t, int64(0), c.proc.Mp().CurrNB())
 }
 
diff --git a/pkg/sql/colexec/offset/offset_test.go b/pkg/sql/colexec/offset/offset_test.go
index bd276be2a..8cf98adea 100644
--- a/pkg/sql/colexec/offset/offset_test.go
+++ b/pkg/sql/colexec/offset/offset_test.go
@@ -109,7 +109,6 @@ func TestOffset(t *testing.T) {
 		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = nil
 		_, _ = Call(0, tc.proc, tc.arg)
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/order/order_test.go b/pkg/sql/colexec/order/order_test.go
index cbfe0e942..8f3281401 100644
--- a/pkg/sql/colexec/order/order_test.go
+++ b/pkg/sql/colexec/order/order_test.go
@@ -84,7 +84,6 @@ func TestOrder(t *testing.T) {
 		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = nil
 		_, _ = Call(0, tc.proc, tc.arg)
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/output/output_test.go b/pkg/sql/colexec/output/output_test.go
index 772d2932b..123df09b3 100644
--- a/pkg/sql/colexec/output/output_test.go
+++ b/pkg/sql/colexec/output/output_test.go
@@ -90,7 +90,6 @@ func TestOutput(t *testing.T) {
 		tc.proc.Reg.InputBatch = nil
 		_, err = Call(0, tc.proc, tc.arg)
 		require.NoError(t, err)
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/product/product_test.go b/pkg/sql/colexec/product/product_test.go
index 3c233e3e9..30f067c8d 100644
--- a/pkg/sql/colexec/product/product_test.go
+++ b/pkg/sql/colexec/product/product_test.go
@@ -87,7 +87,6 @@ func TestProduct(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/projection/projection_test.go b/pkg/sql/colexec/projection/projection_test.go
index 359f7b160..d9a91308d 100644
--- a/pkg/sql/colexec/projection/projection_test.go
+++ b/pkg/sql/colexec/projection/projection_test.go
@@ -90,7 +90,6 @@ func TestProjection(t *testing.T) {
 		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = nil
 		_, _ = Call(0, tc.proc, tc.arg)
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/semi/join_test.go b/pkg/sql/colexec/semi/join_test.go
index 44e42fc4f..58caa7850 100644
--- a/pkg/sql/colexec/semi/join_test.go
+++ b/pkg/sql/colexec/semi/join_test.go
@@ -97,7 +97,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/single/join_test.go b/pkg/sql/colexec/single/join_test.go
index df92617e7..e976a2c75 100644
--- a/pkg/sql/colexec/single/join_test.go
+++ b/pkg/sql/colexec/single/join_test.go
@@ -89,7 +89,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 	for _, tc := range tcs {
@@ -108,7 +107,6 @@ func TestJoin(t *testing.T) {
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp())
 		}
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/top/top_test.go b/pkg/sql/colexec/top/top_test.go
index ee8458439..3e91ade12 100644
--- a/pkg/sql/colexec/top/top_test.go
+++ b/pkg/sql/colexec/top/top_test.go
@@ -82,7 +82,6 @@ func TestTop(t *testing.T) {
 		}
 		tc.proc.Reg.InputBatch = nil
 		_, _ = Call(0, tc.proc, tc.arg)
-		tc.arg.Free(tc.proc, false)
 		require.Equal(t, int64(0), tc.proc.Mp().CurrNB())
 	}
 }
diff --git a/pkg/sql/colexec/unnest/unnest_test.go b/pkg/sql/colexec/unnest/unnest_test.go
index 75fcd351a..526bb1cf5 100644
--- a/pkg/sql/colexec/unnest/unnest_test.go
+++ b/pkg/sql/colexec/unnest/unnest_test.go
@@ -143,7 +143,6 @@ func TestString(t *testing.T) {
 
 func TestUnnest(t *testing.T) {
 	for i, ut := range utc {
-		nb0 := ut.proc.Mp().CurrNB()
 		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
 			err := Prepare(ut.proc, ut.arg)
 			require.Nil(t, err)
@@ -154,7 +153,6 @@ func TestUnnest(t *testing.T) {
 				require.Nil(t, err)
 				require.False(t, end)
 				require.NotNil(t, ut.proc.InputBatch())
-				ut.proc.Reg.InputBatch.Clean(ut.proc.Mp())
 				ut.proc.SetInputBatch(nil)
 				end, err = Call(0, ut.proc, ut.arg)
 				require.Nil(t, err)
@@ -171,24 +169,17 @@ func TestUnnest(t *testing.T) {
 				require.False(t, end)
 				require.Nil(t, err)
 				require.NotNil(t, ut.proc.InputBatch())
-				ut.proc.Reg.InputBatch.Clean(ut.proc.Mp())
 			}
-			ut.proc.SetInputBatch(nil)
+			ut.proc.Reg.InputBatch = nil
 			end, err := Call(0, ut.proc, ut.arg)
 			require.Nil(t, err)
 			require.True(t, end)
 		})
-		if ut.proc.InputBatch() != nil {
-			ut.proc.Reg.InputBatch.Clean(ut.proc.Mp())
-		}
-		ut.arg.Free(ut.proc, false)
-		require.Equal(t, nb0, ut.proc.Mp().CurrNB())
 	}
 }
 
 func makeTestBatch1(json string, proc *process.Process) (*batch.Batch, error) {
 	bat := batch.New(true, []string{"src"})
-	bat.Cnt = 1
 	bat.Vecs[0] = vector.New(types.Type{
 		Oid: types.T_varchar,
 	})
@@ -202,7 +193,6 @@ func makeTestBatch1(json string, proc *process.Process) (*batch.Batch, error) {
 
 func makeTestBatch2(jsons []string, proc *process.Process) (*batch.Batch, error) {
 	bat := batch.New(true, []string{"a"})
-	bat.Cnt = 1
 	for i := range bat.Vecs {
 		bat.Vecs[i] = vector.New(types.Type{
 			Oid:   types.T_json,