Add analyze information (#4070)

pkg/container/batch/batch.go

@@ -234,6 +234,15 @@ func (bat *Batch) Shuffle(sels []int64, m *mheap.Mheap) error {
 	return nil
 }
 
+func (bat *Batch) Size() int {
+	var size int
+
+	for _, vec := range bat.Vecs {
+		size += len(vec.Data)
+	}
+	return size
+}
+
 func (bat *Batch) Length() int {
 	return len(bat.Zs)
 }

pkg/sql/colexec/complement/complement.go

@@ -73,7 +73,7 @@ func Prepare(proc *process.Process, arg interface{}) error {
 	return nil
 }
 
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	ap := arg.(*Argument)
 	ctr := ap.ctr
 	for {

pkg/sql/colexec/complement/complement_test.go

@@ -235,7 +235,7 @@ func TestComplement(t *testing.T) {
 		tc.proc.Reg.MergeReceivers[1].Ch <- &batch.Batch{}
 		tc.proc.Reg.MergeReceivers[1].Ch <- nil
 		for {
-			if ok, err := Call(tc.proc, tc.arg); ok || err != nil {
+			if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
 				break
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
@@ -282,7 +282,7 @@ func BenchmarkComplement(b *testing.B) {
 			tc.proc.Reg.MergeReceivers[1].Ch <- &batch.Batch{}
 			tc.proc.Reg.MergeReceivers[1].Ch <- nil
 			for {
-				if ok, err := Call(tc.proc, tc.arg); ok || err != nil {
+				if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
 					break
 				}
 				tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)

pkg/sql/colexec/connector/connector.go

@@ -29,7 +29,7 @@ func Prepare(_ *process.Process, _ interface{}) error {
 	return nil
 }
 
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	ap := arg.(*Argument)
 	reg := ap.Reg
 	bat := proc.Reg.InputBatch

pkg/sql/colexec/connector/connector_test.go

@@ -80,11 +80,11 @@ func TestConnector(t *testing.T) {
 				}
 			}
 		}
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = &batch.Batch{}
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = nil
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		for {
 			bat := <-tc.arg.Reg.Ch
 			if bat == nil {

pkg/sql/colexec/deletion/deletion.go

@@ -16,9 +16,10 @@ package deletion
 
 import (
 	"bytes"
+	"sync/atomic"
+
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
-	"sync/atomic"
 )
 
 func String(arg interface{}, buf *bytes.Buffer) {
@@ -29,7 +30,7 @@ func Prepare(_ *process.Process, _ interface{}) error {
 	return nil
 }
 
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	p := arg.(*Argument)
 	bat := proc.Reg.InputBatch
 	if bat == nil || len(bat.Zs) == 0 {

pkg/sql/colexec/dispatch/dispatch.go

@@ -32,7 +32,7 @@ func Prepare(_ *process.Process, arg interface{}) error {
 	return nil
 }
 
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	ap := arg.(*Argument)
 	bat := proc.Reg.InputBatch
 	if bat == nil {

pkg/sql/colexec/dispatch/dispatch_test.go

@@ -81,11 +81,11 @@ func TestDispatch(t *testing.T) {
 				}
 			}
 		}
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = &batch.Batch{}
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = nil
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		for {
 			bat := <-tc.arg.Regs[0].Ch
 			if bat == nil {

pkg/sql/colexec/group/group.go

@@ -56,7 +56,7 @@ func Prepare(_ *process.Process, arg interface{}) error {
 	return nil
 }
 
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	ap := arg.(*Argument)
 	if len(ap.Exprs) == 0 {
 		return ap.ctr.process(ap, proc)

pkg/sql/colexec/group/group_test.go

@@ -109,22 +109,22 @@ func TestGroup(t *testing.T) {
 		err := Prepare(tc.proc, tc.arg)
 		require.NoError(t, err)
 		tc.proc.Reg.InputBatch = newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
-		_, err = Call(tc.proc, tc.arg)
+		_, err = Call(0, tc.proc, tc.arg)
 		require.NoError(t, err)
 		tc.proc.Reg.InputBatch = newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
-		_, err = Call(tc.proc, tc.arg)
+		_, err = Call(0, tc.proc, tc.arg)
 		require.NoError(t, err)
 		tc.proc.Reg.InputBatch = &batch.Batch{}
-		_, err = Call(tc.proc, tc.arg)
+		_, err = Call(0, tc.proc, tc.arg)
 		require.NoError(t, err)
 		tc.proc.Reg.InputBatch = nil
-		_, err = Call(tc.proc, tc.arg)
+		_, err = Call(0, tc.proc, tc.arg)
 		require.NoError(t, err)
 		if tc.proc.Reg.InputBatch != nil {
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
 		}
 		tc.proc.Reg.InputBatch = nil
-		_, err = Call(tc.proc, tc.arg)
+		_, err = Call(0, tc.proc, tc.arg)
 		require.NoError(t, err)
 		require.Equal(t, int64(0), mheap.Size(tc.proc.Mp))
 	}
@@ -143,16 +143,16 @@ func BenchmarkGroup(b *testing.B) {
 			err := Prepare(tc.proc, tc.arg)
 			require.NoError(t, err)
 			tc.proc.Reg.InputBatch = newBatch(t, tc.flgs, tc.types, tc.proc, BenchmarkRows)
-			_, err = Call(tc.proc, tc.arg)
+			_, err = Call(0, tc.proc, tc.arg)
 			require.NoError(t, err)
 			tc.proc.Reg.InputBatch = newBatch(t, tc.flgs, tc.types, tc.proc, BenchmarkRows)
-			_, err = Call(tc.proc, tc.arg)
+			_, err = Call(0, tc.proc, tc.arg)
 			require.NoError(t, err)
 			tc.proc.Reg.InputBatch = &batch.Batch{}
-			_, err = Call(tc.proc, tc.arg)
+			_, err = Call(0, tc.proc, tc.arg)
 			require.NoError(t, err)
 			tc.proc.Reg.InputBatch = nil
-			_, err = Call(tc.proc, tc.arg)
+			_, err = Call(0, tc.proc, tc.arg)
 			require.NoError(t, err)
 			if tc.proc.Reg.InputBatch != nil {
 				tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)

pkg/sql/colexec/insert/insert.go

@@ -17,6 +17,7 @@ package insert
 import (
 	"bytes"
 	"fmt"
+
 	"github.com/matrixorigin/matrixone/pkg/container/nulls"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
@@ -42,7 +43,7 @@ func Prepare(_ *process.Process, _ interface{}) error {
 	return nil
 }
 
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	n := arg.(*Argument)
 	bat := proc.Reg.InputBatch
 	if bat == nil {

pkg/sql/colexec/insert/insert_test.go

@@ -66,7 +66,7 @@ func TestInsertOperator(t *testing.T) {
 		},
 	}
 	proc.Reg.InputBatch = batch1
-	_, err := Call(proc, &argument1)
+	_, err := Call(0, proc, &argument1)
 	require.NoError(t, err)
 	println(argument1.TargetTable.(*mockRelation).result.Vecs)
 	{
@@ -104,6 +104,6 @@ func TestInsertOperator(t *testing.T) {
 		},
 	}
 	proc.Reg.InputBatch = batch2
-	_, err2 := Call(proc, &argument2)
+	_, err2 := Call(0, proc, &argument2)
 	require.Errorf(t, err2, "should return error when insert null into primary key column")
 }

pkg/sql/colexec/join/join.go

@@ -83,7 +83,7 @@ func Prepare(proc *process.Process, arg interface{}) error {
 	return nil
 }
 
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	ap := arg.(*Argument)
 	ctr := ap.ctr
 	for {

pkg/sql/colexec/join/join_test.go

@@ -236,7 +236,7 @@ func TestJoin(t *testing.T) {
 		tc.proc.Reg.MergeReceivers[1].Ch <- &batch.Batch{}
 		tc.proc.Reg.MergeReceivers[1].Ch <- nil
 		for {
-			if ok, err := Call(tc.proc, tc.arg); ok || err != nil {
+			if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
 				break
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
@@ -283,7 +283,7 @@ func BenchmarkJoin(b *testing.B) {
 			tc.proc.Reg.MergeReceivers[1].Ch <- &batch.Batch{}
 			tc.proc.Reg.MergeReceivers[1].Ch <- nil
 			for {
-				if ok, err := Call(tc.proc, tc.arg); ok || err != nil {
+				if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
 					break
 				}
 				tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)

pkg/sql/colexec/left/join.go

@@ -83,7 +83,7 @@ func Prepare(proc *process.Process, arg interface{}) error {
 	return nil
 }
 
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	ap := arg.(*Argument)
 	ctr := ap.ctr
 	for {

pkg/sql/colexec/left/join_test.go

@@ -236,7 +236,7 @@ func TestJoin(t *testing.T) {
 		tc.proc.Reg.MergeReceivers[1].Ch <- &batch.Batch{}
 		tc.proc.Reg.MergeReceivers[1].Ch <- nil
 		for {
-			if ok, err := Call(tc.proc, tc.arg); ok || err != nil {
+			if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
 				break
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
@@ -283,7 +283,7 @@ func BenchmarkJoin(b *testing.B) {
 			tc.proc.Reg.MergeReceivers[1].Ch <- &batch.Batch{}
 			tc.proc.Reg.MergeReceivers[1].Ch <- nil
 			for {
-				if ok, err := Call(tc.proc, tc.arg); ok || err != nil {
+				if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
 					break
 				}
 				tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)

pkg/sql/colexec/limit/limit.go

@@ -32,7 +32,7 @@ func Prepare(_ *process.Process, _ interface{}) error {
 }
 
 // returning only the first n tuples from its input
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	bat := proc.Reg.InputBatch
 	if bat == nil {
 		return true, nil

pkg/sql/colexec/limit/limit_test.go

@@ -100,19 +100,19 @@ func TestLimit(t *testing.T) {
 		err := Prepare(tc.proc, tc.arg)
 		require.NoError(t, err)
 		tc.proc.Reg.InputBatch = newBatch(t, tc.types, tc.proc, Rows)
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		if tc.proc.Reg.InputBatch != nil {
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
 		}
 		tc.proc.Reg.InputBatch = newBatch(t, tc.types, tc.proc, Rows)
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		if tc.proc.Reg.InputBatch != nil {
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
 		}
 		tc.proc.Reg.InputBatch = &batch.Batch{}
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		tc.proc.Reg.InputBatch = nil
-		_, _ = Call(tc.proc, tc.arg)
+		_, _ = Call(0, tc.proc, tc.arg)
 		require.Equal(t, int64(0), mheap.Size(tc.proc.Mp))
 	}
 }
@@ -139,14 +139,14 @@ func BenchmarkLimit(b *testing.B) {
 			err := Prepare(tc.proc, tc.arg)
 			require.NoError(t, err)
 			tc.proc.Reg.InputBatch = newBatch(t, tc.types, tc.proc, BenchmarkRows)
-			_, _ = Call(tc.proc, tc.arg)
+			_, _ = Call(0, tc.proc, tc.arg)
 			if tc.proc.Reg.InputBatch != nil {
 				tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
 			}
 			tc.proc.Reg.InputBatch = &batch.Batch{}
-			_, _ = Call(tc.proc, tc.arg)
+			_, _ = Call(0, tc.proc, tc.arg)
 			tc.proc.Reg.InputBatch = nil
-			_, _ = Call(tc.proc, tc.arg)
+			_, _ = Call(0, tc.proc, tc.arg)
 		}
 	}
 }

pkg/sql/colexec/loopcomplement/complement.go

@@ -34,7 +34,7 @@ func Prepare(proc *process.Process, arg interface{}) error {
 	return nil
 }
 
-func Call(proc *process.Process, arg interface{}) (bool, error) {
+func Call(_ int, proc *process.Process, arg interface{}) (bool, error) {
 	ap := arg.(*Argument)
 	ctr := ap.ctr
 	for {

pkg/sql/colexec/loopcomplement/complement_test.go

@@ -87,7 +87,7 @@ func TestJoin(t *testing.T) {
 		tc.proc.Reg.MergeReceivers[1].Ch <- &batch.Batch{}
 		tc.proc.Reg.MergeReceivers[1].Ch <- nil
 		for {
-			if ok, err := Call(tc.proc, tc.arg); ok || err != nil {
+			if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
 				break
 			}
 			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
@@ -118,7 +118,7 @@ func BenchmarkJoin(b *testing.B) {
 			tc.proc.Reg.MergeReceivers[1].Ch <- &batch.Batch{}
 			tc.proc.Reg.MergeReceivers[1].Ch <- nil
 			for {
-				if ok, err := Call(tc.proc, tc.arg); ok || err != nil {
+				if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
 					break
 				}
 				tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)