AggSpill 3: Save

Now we have something that compiles, but there will be
tons of bugs ...

Author: fengttt

pkg/container/batch/batch.go

@@ -18,13 +18,13 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 
 	"github.com/matrixorigin/matrixone/pkg/common/bitmap"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
-	"github.com/matrixorigin/matrixone/pkg/sql/colexec/aggexec"
 )
 
 func New(attrs []string) *Batch {
@@ -145,37 +145,9 @@ func (bat *Batch) MarshalBinaryWithBuffer(w *bytes.Buffer) ([]byte, error) {
 		}
 	}
 
-	// ExtraBuf1
-	size := int32(len(bat.ExtraBuf1))
-	w.Write(types.EncodeInt32(&size))
-	if size > 0 {
-		w.Write(bat.ExtraBuf1)
-	}
-
-	// ExtraBuf2
-	size = int32(len(bat.ExtraBuf2))
-	w.Write(types.EncodeInt32(&size))
-	if size > 0 {
-		w.Write(bat.ExtraBuf2)
-	}
-
-	// AggInfos
-	aggInfos := make([][]byte, len(bat.Aggs))
-	for i, exec := range bat.Aggs {
-		data, err := aggexec.MarshalAggFuncExec(exec)
-		if err != nil {
-			return nil, err
-		}
-		aggInfos[i] = data
-	}
-
-	l = int32(len(aggInfos))
-	w.Write(types.EncodeInt32(&l))
-	for i := 0; i < int(l); i++ {
-		size := int32(len(aggInfos[i]))
-		w.Write(types.EncodeInt32(&size))
-		w.Write(aggInfos[i])
-	}
+	// ExtraBuf1 and ExtraBuf2
+	types.WriteSizeBytes(bat.ExtraBuf1, w)
+	types.WriteSizeBytes(bat.ExtraBuf2, w)
 
 	w.Write(types.EncodeInt32(&bat.Recursive))
 	w.Write(types.EncodeInt32(&bat.ShuffleIDX))
@@ -247,33 +219,75 @@ func (bat *Batch) UnmarshalBinaryWithAnyMp(data []byte, mp *mpool.MPool) (err er
 	bat.ExtraBuf2 = append(bat.ExtraBuf2, data[:l]...)
 	data = data[l:]
 
-	l = types.DecodeInt32(data[:4])
-	aggs := make([][]byte, l)
-
-	data = data[4:]
-	for i := 0; i < int(l); i++ {
-		size := types.DecodeInt32(data[:4])
-		data = data[4:]
-		aggs[i] = data[:size]
-		data = data[size:]
-	}
-
 	bat.Recursive = types.DecodeInt32(data[:4])
 	data = data[4:]
 	bat.ShuffleIDX = types.DecodeInt32(data[:4])
+	return nil
+}
 
-	if len(aggs) > 0 {
-		bat.Aggs = make([]aggexec.AggFuncExec, len(aggs))
-		var aggMemoryManager aggexec.AggMemoryManager = nil
-		if mp != nil {
-			aggMemoryManager = aggexec.NewSimpleAggMemoryManager(mp)
+func (bat *Batch) UnmarshalFromReader(r io.Reader, mp *mpool.MPool) (err error) {
+	i64, err := types.ReadInt64(r)
+	if err != nil {
+		return err
+	}
+	bat.rowCount = int(i64)
+
+	l, err := types.ReadInt32AsInt(r)
+	if l != len(bat.Vecs) {
+		if len(bat.Vecs) > 0 {
+			bat.Clean(mp)
 		}
-		for i, info := range aggs {
-			if bat.Aggs[i], err = aggexec.UnmarshalAggFuncExec(aggMemoryManager, info); err != nil {
-				return err
+		bat.Vecs = make([]*vector.Vector, l)
+		for i := range bat.Vecs {
+			if bat.offHeap {
+				bat.Vecs[i] = vector.NewOffHeapVec()
+			} else {
+				bat.Vecs[i] = vector.NewVecFromReuse()
 			}
 		}
 	}
+	vecs := bat.Vecs
+
+	for i := 0; i < l; i++ {
+		_, bs, err := types.ReadSizeBytes(r, nil, false)
+		if err != nil {
+			return err
+		}
+		if err := vecs[i].UnmarshalWithReader(bytes.NewReader(bs), nil); err != nil {
+			return err
+		}
+	}
+
+	l, err = types.ReadInt32AsInt(r)
+	if err != nil {
+		return err
+	}
+	if l != len(bat.Attrs) {
+		bat.Attrs = make([]string, l)
+	}
+
+	for i := 0; i < int(l); i++ {
+		_, bs, err := types.ReadSizeBytes(r, nil, false)
+		if err != nil {
+			return err
+		}
+		bat.Attrs[i] = string(bs)
+	}
+
+	// ExtraBuf1
+	if _, bat.ExtraBuf1, err = types.ReadSizeBytes(r, nil, false); err != nil {
+		return err
+	}
+	if _, bat.ExtraBuf2, err = types.ReadSizeBytes(r, nil, false); err != nil {
+		return err
+	}
+
+	if bat.Recursive, err = types.ReadInt32(r); err != nil {
+		return err
+	}
+	if bat.ShuffleIDX, err = types.ReadInt32(r); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -437,14 +451,11 @@ func (bat *Batch) Clean(m *mpool.MPool) {
 			vec.Free(m)
 		}
 	}
-	for _, agg := range bat.Aggs {
-		if agg != nil {
-			agg.Free()
-		}
-	}
-	bat.Aggs = nil
+
 	bat.Vecs = nil
 	bat.Attrs = nil
+	bat.ExtraBuf1 = nil
+	bat.ExtraBuf2 = nil
 	bat.SetRowCount(0)
 }
 
@@ -639,7 +650,7 @@ func (bat *Batch) ReplaceVector(oldVec *vector.Vector, newVec *vector.Vector, st
 }
 
 func (bat *Batch) IsEmpty() bool {
-	return bat.rowCount == 0 && len(bat.Aggs) == 0
+	return bat.rowCount == 0
 }
 
 func (bat *Batch) IsDone() bool {

pkg/container/batch/batch_test.go

@@ -149,8 +149,6 @@ func newBatch(ts []types.Type, rows int) *Batch {
 	bat.ExtraBuf2 = []byte("extra buf 2")
 
 	aggexec.RegisterGroupConcatAgg(0, ",")
-	agg0, _ := aggexec.MakeAgg(aggexec.NewSimpleAggMemoryManager(mp), 0, false, []types.Type{types.T_varchar.ToType()}...)
-	bat.Aggs = []aggexec.AggFuncExec{agg0}
 	bat.Attrs = []string{"1"}
 	return bat
 }

pkg/container/batch/types.go

@@ -16,7 +16,6 @@ package batch
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
-	"github.com/matrixorigin/matrixone/pkg/sql/colexec/aggexec"
 )
 
 // special batch that will never been free.
@@ -52,12 +51,6 @@ type Batch struct {
 	// Vecs col data
 	Vecs []*vector.Vector
 
-	// We really want to put all data through vectors, but, the
-	// Aggs is so f**king insane, so keep it, and add two extra
-	// buffers for sane persons to use.
-	//
-	// XXX MUST REMOVE THIS
-	Aggs      []aggexec.AggFuncExec
 	ExtraBuf1 []byte
 	ExtraBuf2 []byte
 

pkg/container/pSpool/buffer.go

@@ -15,10 +15,11 @@
 package pSpool
 
 import (
+	"sync"
+
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
-	"sync"
 )
 
 type spoolBuffer struct {
@@ -78,12 +79,6 @@ func (b *spoolBuffer) putCacheID(mp *mpool.MPool, id uint32, bat *batch.Batch) {
 	bat.Attrs = bat.Attrs[:0]
 	bat.SetRowCount(0)
 
-	// we won't reuse the aggregation's memories now.
-	for i := range bat.Aggs {
-		bat.Aggs[i].Free()
-	}
-	bat.Aggs = nil
-
 	// put id into free list.
 	b.Lock()
 	b.readyToUse = b.readyToUse[:len(b.readyToUse)+1]

pkg/container/pSpool/copy.go

@@ -15,10 +15,11 @@
 package pSpool
 
 import (
+	"math"
+
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
-	"math"
 )
 
 // cachedBatch is just like the cachedVectorPool in the original code,
@@ -131,9 +132,6 @@ func (cb *cachedBatch) GetCopiedBatch(
 		}
 	}
 
-	dst.Aggs = src.Aggs
-	src.Aggs = nil
-
 	// set row count.
 	dst.SetRowCount(src.RowCount())
 

pkg/container/types/encoding.go

@@ -25,6 +25,7 @@ import (
 	"unsafe"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/common/util"
 	"github.com/matrixorigin/matrixone/pkg/container/bytejson"
 )
@@ -559,8 +560,100 @@ func WriteSizeBytes(bs []byte, w io.Writer) error {
 	if _, err := w.Write(EncodeInt32(&sz)); err != nil {
 		return err
 	}
-	if _, err := w.Write(bs); err != nil {
-		return err
+	if sz > 0 {
+		if _, err := w.Write(bs); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func ReadInt64(r io.Reader) (int64, error) {
+	buf := make([]byte, 8)
+	if _, err := io.ReadFull(r, buf); err != nil {
+		return 0, err
 	}
+	return DecodeInt64(buf), nil
+}
+
+func WriteInt64(w io.Writer, v int64) error {
+	w.Write(EncodeInt64(&v))
 	return nil
 }
+
+func ReadBool(r io.Reader) (bool, error) {
+	buf := make([]byte, 1)
+	if _, err := io.ReadFull(r, buf); err != nil {
+		return false, err
+	}
+	return DecodeBool(buf), nil
+}
+
+func ReadInt32(r io.Reader) (int32, error) {
+	buf := make([]byte, 4)
+	if _, err := io.ReadFull(r, buf); err != nil {
+		return 0, err
+	}
+	return DecodeInt32(buf), nil
+}
+
+func WriteInt32(w io.Writer, v int32) error {
+	w.Write(EncodeInt32(&v))
+	return nil
+}
+
+func ReadInt32AsInt(r io.Reader) (int, error) {
+	buf := make([]byte, 4)
+	if _, err := io.ReadFull(r, buf); err != nil {
+		return 0, err
+	}
+	return int(DecodeInt32(buf)), nil
+}
+
+func ReadByte(r io.Reader) (byte, error) {
+	buf := make([]byte, 1)
+	if _, err := io.ReadFull(r, buf); err != nil {
+		return 0, err
+	}
+	return buf[0], nil
+}
+
+func ReadByteAsInt(r io.Reader) (int, error) {
+	buf := make([]byte, 1)
+	if _, err := io.ReadFull(r, buf); err != nil {
+		return 0, err
+	}
+	return int(buf[0]), nil
+}
+
+func ReadType(r io.Reader) (Type, error) {
+	buf := make([]byte, TSize)
+	if _, err := io.ReadFull(r, buf); err != nil {
+		return Type{}, err
+	}
+	return DecodeType(buf), nil
+}
+
+func ReadSizeBytes(r io.Reader, mp *mpool.MPool, offHeap bool) (int32, []byte, error) {
+	sz, err := ReadInt32(r)
+	if err != nil {
+		return 0, nil, err
+	}
+	if sz > 0 {
+		var bs []byte
+		if mp != nil {
+			bs, err = mp.Alloc(int(sz), offHeap)
+			if err != nil {
+				return 0, nil, err
+			}
+		} else {
+			bs = make([]byte, sz)
+		}
+
+		if _, err := io.ReadFull(r, bs); err != nil {
+			return 0, nil, err
+		}
+		return sz, bs, nil
+	}
+	return sz, nil, nil
+}