// Licensed to the LF AI & Data foundation under one // or more contributor license agreements. See the NOTICE file // distributed with this work for additional information // regarding copyright ownership. The ASF licenses this file // to you 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 json import ( "context" "encoding/json" "io" "math" "strings" "testing" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/suite" "github.com/milvus-io/milvus-proto/go-api/v2/commonpb" "github.com/milvus-io/milvus-proto/go-api/v2/schemapb" "github.com/milvus-io/milvus/internal/mocks" "github.com/milvus-io/milvus/internal/storage" "github.com/milvus-io/milvus/internal/util/nullutil" "github.com/milvus-io/milvus/internal/util/testutil" "github.com/milvus-io/milvus/pkg/common" "github.com/milvus-io/milvus/pkg/util/paramtable" ) type ReaderSuite struct { suite.Suite numRows int pkDataType schemapb.DataType vecDataType schemapb.DataType } func (suite *ReaderSuite) SetupSuite() { paramtable.Get().Init(paramtable.NewBaseTable()) } func (suite *ReaderSuite) SetupTest() { // default suite params suite.numRows = 100 suite.pkDataType = schemapb.DataType_Int64 suite.vecDataType = schemapb.DataType_FloatVector } func (suite *ReaderSuite) run(dataType schemapb.DataType, elemType schemapb.DataType, nullable bool) { schema := &schemapb.CollectionSchema{ Fields: []*schemapb.FieldSchema{ { FieldID: 100, Name: "pk", IsPrimaryKey: true, DataType: suite.pkDataType, TypeParams: []*commonpb.KeyValuePair{ { Key: common.MaxLengthKey, Value: "128", }, }, }, { FieldID: 101, Name: "vec", DataType: suite.vecDataType, TypeParams: []*commonpb.KeyValuePair{ { Key: common.DimKey, Value: "8", }, }, }, { FieldID: 102, Name: dataType.String(), DataType: dataType, ElementType: elemType, TypeParams: []*commonpb.KeyValuePair{ { Key: common.MaxLengthKey, Value: "128", }, { Key: common.MaxCapacityKey, Value: "128", }, }, Nullable: nullable, }, }, } if dataType == schemapb.DataType_VarChar { // Add a BM25 function if data type is VarChar schema.Fields = append(schema.Fields, &schemapb.FieldSchema{ FieldID: 103, Name: "sparse", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true, }) schema.Functions = append(schema.Functions, &schemapb.FunctionSchema{ Id: 1000, Name: "bm25", Type: schemapb.FunctionType_BM25, InputFieldIds: []int64{102}, InputFieldNames: []string{dataType.String()}, OutputFieldIds: []int64{103}, OutputFieldNames: []string{"sparse"}, }) } insertData, err := testutil.CreateInsertData(schema, suite.numRows) suite.NoError(err) rows, err := testutil.CreateInsertDataRowsForJSON(schema, insertData) suite.NoError(err) jsonBytes, err := json.Marshal(rows) suite.NoError(err) type mockReader struct { io.Reader io.Closer io.ReaderAt io.Seeker } cm := mocks.NewChunkManager(suite.T()) cm.EXPECT().Reader(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, s string) (storage.FileReader, error) { r := &mockReader{Reader: strings.NewReader(string(jsonBytes))} return r, nil }) reader, err := NewReader(context.Background(), cm, schema, "mockPath", math.MaxInt) suite.NoError(err) checkFn := func(actualInsertData *storage.InsertData, offsetBegin, expectRows int) { expectInsertData := insertData for fieldID, data := range actualInsertData.Data { suite.Equal(expectRows, data.RowNum()) for i := 0; i < expectRows; i++ { expect := expectInsertData.Data[fieldID].GetRow(i + offsetBegin) actual := data.GetRow(i) suite.Equal(expect, actual) } } } res, err := reader.Read() suite.NoError(err) checkFn(res, 0, suite.numRows) } func (suite *ReaderSuite) runWithDefaultValue(dataType schemapb.DataType, elemType schemapb.DataType) { schema := &schemapb.CollectionSchema{ Fields: []*schemapb.FieldSchema{ { FieldID: 100, Name: "pk", IsPrimaryKey: true, DataType: suite.pkDataType, TypeParams: []*commonpb.KeyValuePair{ { Key: common.MaxLengthKey, Value: "128", }, }, }, { FieldID: 101, Name: "vec", DataType: suite.vecDataType, TypeParams: []*commonpb.KeyValuePair{ { Key: common.DimKey, Value: "8", }, }, }, }, } // here always set nullable==true just for test, insertData will store validData only nullable==true // jsonBytes Marshal from rows, if expectInsertData is nulls and set default value // actualData will be default_value fieldSchema, err := testutil.CreateFieldWithDefaultValue(dataType, 102, true) suite.NoError(err) schema.Fields = append(schema.Fields, fieldSchema) insertData, err := testutil.CreateInsertData(schema, suite.numRows) suite.NoError(err) rows, err := testutil.CreateInsertDataRowsForJSON(schema, insertData) suite.NoError(err) jsonBytes, err := json.Marshal(rows) suite.NoError(err) type mockReader struct { io.Reader io.Closer io.ReaderAt io.Seeker } cm := mocks.NewChunkManager(suite.T()) cm.EXPECT().Reader(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, s string) (storage.FileReader, error) { r := &mockReader{Reader: strings.NewReader(string(jsonBytes))} return r, nil }) reader, err := NewReader(context.Background(), cm, schema, "mockPath", math.MaxInt) suite.NoError(err) checkFn := func(actualInsertData *storage.InsertData, offsetBegin, expectRows int) { expectInsertData := insertData for fieldID, data := range actualInsertData.Data { suite.Equal(expectRows, data.RowNum()) for i := 0; i < expectRows; i++ { expect := expectInsertData.Data[fieldID].GetRow(i + offsetBegin) actual := data.GetRow(i) if expect == nil { expect, err = nullutil.GetDefaultValue(fieldSchema) suite.NoError(err) } suite.Equal(expect, actual) } } } res, err := reader.Read() suite.NoError(err) checkFn(res, 0, suite.numRows) } func (suite *ReaderSuite) TestReadScalarFields() { suite.run(schemapb.DataType_Bool, schemapb.DataType_None, false) suite.run(schemapb.DataType_Int8, schemapb.DataType_None, false) suite.run(schemapb.DataType_Int16, schemapb.DataType_None, false) suite.run(schemapb.DataType_Int32, schemapb.DataType_None, false) suite.run(schemapb.DataType_Int64, schemapb.DataType_None, false) suite.run(schemapb.DataType_Float, schemapb.DataType_None, false) suite.run(schemapb.DataType_Double, schemapb.DataType_None, false) suite.run(schemapb.DataType_String, schemapb.DataType_None, false) suite.run(schemapb.DataType_VarChar, schemapb.DataType_None, false) suite.run(schemapb.DataType_JSON, schemapb.DataType_None, false) suite.run(schemapb.DataType_Array, schemapb.DataType_Bool, false) suite.run(schemapb.DataType_Array, schemapb.DataType_Int8, false) suite.run(schemapb.DataType_Array, schemapb.DataType_Int16, false) suite.run(schemapb.DataType_Array, schemapb.DataType_Int32, false) suite.run(schemapb.DataType_Array, schemapb.DataType_Int64, false) suite.run(schemapb.DataType_Array, schemapb.DataType_Float, false) suite.run(schemapb.DataType_Array, schemapb.DataType_Double, false) suite.run(schemapb.DataType_Array, schemapb.DataType_String, false) suite.run(schemapb.DataType_Bool, schemapb.DataType_None, true) suite.run(schemapb.DataType_Int8, schemapb.DataType_None, true) suite.run(schemapb.DataType_Int16, schemapb.DataType_None, true) suite.run(schemapb.DataType_Int32, schemapb.DataType_None, true) suite.run(schemapb.DataType_Int64, schemapb.DataType_None, true) suite.run(schemapb.DataType_Float, schemapb.DataType_None, true) suite.run(schemapb.DataType_Double, schemapb.DataType_None, true) suite.run(schemapb.DataType_String, schemapb.DataType_None, true) suite.run(schemapb.DataType_VarChar, schemapb.DataType_None, true) suite.run(schemapb.DataType_JSON, schemapb.DataType_None, true) suite.run(schemapb.DataType_Array, schemapb.DataType_Bool, true) suite.run(schemapb.DataType_Array, schemapb.DataType_Int8, true) suite.run(schemapb.DataType_Array, schemapb.DataType_Int16, true) suite.run(schemapb.DataType_Array, schemapb.DataType_Int32, true) suite.run(schemapb.DataType_Array, schemapb.DataType_Int64, true) suite.run(schemapb.DataType_Array, schemapb.DataType_Float, true) suite.run(schemapb.DataType_Array, schemapb.DataType_Double, true) suite.run(schemapb.DataType_Array, schemapb.DataType_String, true) } func (suite *ReaderSuite) TestReadScalarFieldsWithDefaultValue() { suite.runWithDefaultValue(schemapb.DataType_Bool, schemapb.DataType_None) suite.runWithDefaultValue(schemapb.DataType_Int8, schemapb.DataType_None) suite.runWithDefaultValue(schemapb.DataType_Int16, schemapb.DataType_None) suite.runWithDefaultValue(schemapb.DataType_Int32, schemapb.DataType_None) suite.runWithDefaultValue(schemapb.DataType_Int64, schemapb.DataType_None) suite.runWithDefaultValue(schemapb.DataType_Float, schemapb.DataType_None) suite.runWithDefaultValue(schemapb.DataType_Double, schemapb.DataType_None) suite.runWithDefaultValue(schemapb.DataType_String, schemapb.DataType_None) suite.runWithDefaultValue(schemapb.DataType_VarChar, schemapb.DataType_None) } func (suite *ReaderSuite) TestStringPK() { suite.pkDataType = schemapb.DataType_VarChar suite.run(schemapb.DataType_Int32, schemapb.DataType_None, false) } func (suite *ReaderSuite) TestVector() { suite.vecDataType = schemapb.DataType_BinaryVector suite.run(schemapb.DataType_Int32, schemapb.DataType_None, false) suite.vecDataType = schemapb.DataType_FloatVector suite.run(schemapb.DataType_Int32, schemapb.DataType_None, false) suite.vecDataType = schemapb.DataType_Float16Vector suite.run(schemapb.DataType_Int32, schemapb.DataType_None, false) suite.vecDataType = schemapb.DataType_BFloat16Vector suite.run(schemapb.DataType_Int32, schemapb.DataType_None, false) suite.vecDataType = schemapb.DataType_SparseFloatVector suite.run(schemapb.DataType_Int32, schemapb.DataType_None, false) } func TestUtil(t *testing.T) { suite.Run(t, new(ReaderSuite)) }