fix top container reset (#21208) (#21335)

fix top container reset

Approved by: @badboynt1, @sukki37

Author: ouyuanning

pkg/sql/colexec/top/types.go

@@ -95,15 +95,14 @@ func (top *Top) Release() {
 }
 
 func (top *Top) Reset(proc *process.Process, pipelineFailed bool, err error) {
-	top.ctr.reset()
+	top.ctr.reset(proc)
 }
 
 func (top *Top) Free(proc *process.Process, pipelineFailed bool, err error) {
 	top.ctr.free(proc)
 }
 
-func (ctr *container) reset() {
-
+func (ctr *container) reset(proc *process.Process) {
 	ctr.n = 0
 	ctr.state = 0
 	ctr.sels = nil
@@ -123,9 +122,14 @@ func (ctr *container) reset() {
 	ctr.desc = false
 	ctr.topValueZM = nil
 	if ctr.bat != nil {
-		ctr.bat.CleanOnlyData()
+		ctr.bat.Clean(proc.Mp())
+		ctr.bat = nil
 	}
 
+	if ctr.buildBat != nil {
+		ctr.buildBat.Clean(proc.Mp())
+		ctr.buildBat = nil
+	}
 }
 
 func (ctr *container) free(proc *process.Process) {

pkg/sql/colexec/top/types_test.go

@@ -0,0 +1,51 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package top
+
+import (
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/matrixorigin/matrixone/pkg/vm/process"
+)
+
+func Test_container_reset(t *testing.T) {
+	bat := batch.New([]string{"id"})
+	bat.Vecs[0] = testutil.MakeInt32Vector([]int32{1, 2, 3}, nil)
+	buildBat := batch.New([]string{"id"})
+	buildBat.Vecs[0] = testutil.MakeInt32Vector([]int32{1, 2, 3}, nil)
+
+	proc := &process.Process{
+		Base: &process.BaseProcess{},
+	}
+	proc.SetMPool(testutil.TestUtilMp)
+
+	c := &container{
+		n:                       0,
+		state:                   0,
+		sels:                    nil,
+		poses:                   nil,
+		cmps:                    nil,
+		limit:                   0,
+		limitExecutor:           nil,
+		executorsForOrderColumn: nil,
+		desc:                    false,
+		topValueZM:              nil,
+		bat:                     bat,
+		buildBat:                buildBat,
+	}
+	c.reset(proc)
+}