copr: Support VECTOR for RowFormatv1 (#55356)

ref #54245

pkg/types/vector.go

@@ -131,22 +131,31 @@ func (v VectorFloat32) EstimatedMemUsage() int {
 	return int(unsafe.Sizeof(v)) + len(v.data)
 }
 
-// ZeroCopyDeserializeVectorFloat32 deserializes the byte slice into a vector, without memory copy.
-// Note: b must not be mutated, because this function does zero copy.
-func ZeroCopyDeserializeVectorFloat32(b []byte) (VectorFloat32, []byte, error) {
+// PeekBytesAsVectorFloat32 trys to peek some bytes from b, until
+// we can deserialize a VectorFloat32 from those bytes.
+func PeekBytesAsVectorFloat32(b []byte) (n int, err error) {
 	if len(b) < 4 {
-		return ZeroVectorFloat32, b, errors.Errorf("bad VectorFloat32 value header (len=%d)", len(b))
+		err = errors.Errorf("bad VectorFloat32 value header (len=%d)", len(b))
+		return
 	}
 
 	elements := binary.LittleEndian.Uint32(b)
 	totalDataSize := elements*4 + 4
 	if len(b) < int(totalDataSize) {
-		return ZeroVectorFloat32, b, errors.Errorf("bad VectorFloat32 value (len=%d, expected=%d)", len(b), totalDataSize)
+		err = errors.Errorf("bad VectorFloat32 value (len=%d, expected=%d)", len(b), totalDataSize)
+		return
 	}
+	return int(totalDataSize), nil
+}
 
-	data := b[:totalDataSize]
-	remaining := b[totalDataSize:]
-	return VectorFloat32{data: data}, remaining, nil
+// ZeroCopyDeserializeVectorFloat32 deserializes the byte slice into a vector, without memory copy.
+// Note: b must not be mutated, because this function does zero copy.
+func ZeroCopyDeserializeVectorFloat32(b []byte) (VectorFloat32, []byte, error) {
+	length, err := PeekBytesAsVectorFloat32(b)
+	if err != nil {
+		return ZeroVectorFloat32, b, err
+	}
+	return VectorFloat32{data: b[:length]}, b[length:], nil
 }
 
 // ParseVectorFloat32 parses a string into a vector.

pkg/types/vector_test.go

@@ -131,3 +131,28 @@ func TestVectorCompare(t *testing.T) {
 	require.Equal(t, -1, v1.Compare(v2))
 	require.Equal(t, 1, v2.Compare(v1))
 }
+
+func TestVectorSerialize(t *testing.T) {
+	v1, err := types.ParseVectorFloat32(`[1.1, 2.2, 3.3]`)
+	require.NoError(t, err)
+
+	serialized := v1.SerializeTo(nil)
+	serialized = append(serialized, 0x01, 0x02, 0x03, 0x04)
+
+	v2, remaining, err := types.ZeroCopyDeserializeVectorFloat32(serialized)
+	require.NoError(t, err)
+	require.Len(t, remaining, 4)
+	require.Equal(t, []byte{
+		0x01, 0x02, 0x03, 0x04,
+	}, remaining)
+
+	require.Equal(t, "[1.1,2.2,3.3]", v2.String())
+
+	v3, remaining, err := types.ZeroCopyDeserializeVectorFloat32([]byte{0xF1, 0xFC})
+	require.Error(t, err)
+	require.Len(t, remaining, 2)
+	require.Equal(t, []byte{
+		0xF1, 0xFC,
+	}, remaining)
+	require.True(t, v3.IsZeroValue())
+}

pkg/util/codec/codec.go

@@ -1206,6 +1206,8 @@ func peek(b []byte) (length int, err error) {
 		l, err = peekUvarint(b)
 	case jsonFlag:
 		l, err = types.PeekBytesAsJSON(b)
+	case vectorFloat32Flag:
+		l, err = types.PeekBytesAsVectorFloat32(b)
 	default:
 		return 0, errors.Errorf("invalid encoded key flag %v", flag)
 	}