chore: upgrade kernel record struct

Signed-off-by: Young Xu <[email protected]>

Author: xuthus5

lib/record/column_util.go

@@ -15,6 +15,7 @@
 package record
 
 import (
+	"fmt"
 	"unsafe"
 )
 
@@ -226,3 +227,116 @@ func (cv *ColVal) Marshal(buf []byte) ([]byte, error) {
 	buf = AppendUint32Slice(buf, cv.Offset)
 	return buf, nil
 }
+
+func resize(dst []ColVal, segs int) []ColVal {
+	if cap(dst) < segs {
+		delta := segs - cap(dst)
+		dst = dst[:cap(dst)]
+		dst = append(dst, make([]ColVal, delta)...)
+	}
+	dst = dst[:segs]
+	return dst
+}
+
+func (cv *ColVal) Split(dst []ColVal, maxRows int, refType int) []ColVal {
+	if maxRows <= 0 {
+		panic(fmt.Sprintf("maxRows is %v, must grater than 0", maxRows))
+	}
+	segs := (cv.Len + maxRows - 1) / maxRows
+	dst = resize(dst, segs)
+
+	start, offset, validCount := 0, 0, 0
+	for i := 0; i < segs; i++ {
+		end := start + maxRows
+		if end > cv.Len {
+			end = cv.Len
+		}
+		offset, validCount = dst[i].sliceValAndOffset(cv, start, end, refType, offset)
+		dst[i].sliceBitMap(cv, start, end)
+		dst[i].Len = end - start
+		dst[i].NilCount = dst[i].Len - validCount
+		start = end
+	}
+
+	return dst
+}
+
+func (cv *ColVal) Length() int {
+	return cv.Len
+}
+
+func (cv *ColVal) ValidCount(start, end int) int {
+	validCount := 0
+	if cv.Length()+cv.NilCount == 0 || cv.Len == cv.NilCount {
+		return validCount
+	}
+	if cv.NilCount == 0 {
+		return end - start
+	}
+
+	end += cv.BitMapOffset
+	start += cv.BitMapOffset
+	for i := start; i < end; i++ {
+		if cv.Bitmap[i>>3]&BitMask[i&0x07] != 0 {
+			validCount++
+		}
+	}
+	return validCount
+}
+
+func (cv *ColVal) sliceValAndOffset(srcCol *ColVal, start, end, colType, valOffset int) (offset int, valueValidCount int) {
+	var validCount, endOffset int
+	if colType == FieldTypeInt {
+		validCount = srcCol.ValidCount(start, end)
+		endOffset = valOffset + Int64SizeBytes*validCount
+		cv.Val = srcCol.Val[valOffset:endOffset]
+	} else if colType == FieldTypeFloat {
+		validCount = srcCol.ValidCount(start, end)
+		endOffset = valOffset + Float64SizeBytes*validCount
+		cv.Val = srcCol.Val[valOffset:endOffset]
+	} else if colType == FieldTypeString {
+		offsetStart := srcCol.Offset[start]
+		if end == srcCol.Len {
+			endOffset = len(srcCol.Val)
+		} else {
+			endOffset = int(srcCol.Offset[end])
+		}
+		cv.Val = srcCol.Val[offsetStart:endOffset]
+		cv.reserveValOffset(end - start)
+		for index, pos := 0, start; pos < end; pos++ {
+			cv.Offset[index] = srcCol.Offset[pos] - offsetStart
+			index++
+			bitOffset := srcCol.BitMapOffset + pos
+			if srcCol.Bitmap[bitOffset>>3]&BitMask[bitOffset&0x07] != 0 {
+				validCount++
+			}
+		}
+	} else if colType == FieldTypeBoolean {
+		validCount = srcCol.ValidCount(start, end)
+		endOffset = valOffset + BooleanSizeBytes*validCount
+		cv.Val = srcCol.Val[valOffset:endOffset]
+	} else {
+		panic("error type")
+	}
+	return endOffset, validCount
+}
+
+func (cv *ColVal) reserveValOffset(size int) {
+	if cap(cv.Offset) < size {
+		cv.Offset = make([]uint32, size)
+	}
+	cv.Offset = cv.Offset[:size]
+}
+
+func (cv *ColVal) sliceBitMap(srcCol *ColVal, start, end int) {
+	s := srcCol.BitMapOffset + start
+	offset := s % 8
+	s = s / 8
+	e := (srcCol.BitMapOffset + end) / 8
+	if (srcCol.BitMapOffset+end)%8 != 0 {
+		e++
+	}
+
+	cv.Bitmap = srcCol.Bitmap[s:e]
+	cv.BitMapOffset = offset
+}

lib/record/field.go

@@ -54,6 +54,12 @@ func (f *Field) String() string {
 
 type Schemas []Field
 
+func (sh Schemas) Len() int { return len(sh) }
+
+func (sh Schemas) Swap(i, j int) { sh[i], sh[j] = sh[j], sh[i] }
+
+func (sh Schemas) Less(i, j int) bool { return sh[i].Name < sh[j].Name }
+
 func (sh *Schemas) String() string {
 	sb := strings.Builder{}
 	for _, f := range *sh {

lib/record/record.go

@@ -181,7 +181,6 @@ func (rec *Record) AppendTime(time ...int64) {
 
 func (rec *Record) Marshal(buf []byte) ([]byte, error) {
 	var err error
-	// Schema
 	buf = AppendUint32(buf, uint32(len(rec.Schema)))
 	for i := 0; i < len(rec.Schema); i++ {
 		buf = AppendUint32(buf, uint32(rec.Schema[i].Size()))
@@ -191,7 +190,6 @@ func (rec *Record) Marshal(buf []byte) ([]byte, error) {
 		}
 	}
 
-	// ColVal
 	buf = AppendUint32(buf, uint32(len(rec.ColVals)))
 	for i := 0; i < len(rec.ColVals); i++ {
 		buf = AppendUint32(buf, uint32(rec.ColVals[i].Size()))
@@ -202,3 +200,34 @@ func (rec *Record) Marshal(buf []byte) ([]byte, error) {
 	}
 	return buf, nil
 }
+
+func (rec *Record) Split(dst []Record, maxRows int) []Record {
+	rows := rec.RowNums()
+	segs := (rows + maxRows - 1) / maxRows
+	if cap(dst) < segs {
+		delta := segs - cap(dst)
+		dst = dst[:cap(dst)]
+		dst = append(dst, make([]Record, delta)...)
+	}
+	dst = dst[:segs]
+
+	if segs == 1 {
+		dst[0] = *rec
+		return dst
+	}
+
+	for i := range dst {
+		dst[i].Schema = append(dst[i].Schema[:0], rec.Schema...)
+		dst[i].ColVals = resize(dst[i].ColVals, rec.Schema.Len())
+	}
+
+	for i := range rec.Schema {
+		col := rec.Column(i)
+		dstCol := col.Split(nil, maxRows, rec.Schema[i].Type)
+		for j := range dstCol {
+			dst[j].ColVals[i] = dstCol[j]
+		}
+	}
+
+	return dst
+}