Repository: milvus-io/milvus-sdk-go
Pull request #456 (2023-05-20 06:46:38)

Adapt dynamic field and support insert by rows
See also milvus-io/milvus#24232
/kind enhancement

sre-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/milvus-io/milvus-sdk-go/pull/456#" title="Author self-approved">congqixia</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=milvus-io%2Fmilvus-sdk-go).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/milvus-io/milvus-sdk-go/blob/master/OWNERS)~~ [congqixia]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/456?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) Report
> Merging [#456](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/456?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) (4aac7eb) into [master](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/commit/c8aa151d65a26d596cd143e5a1daf329f904f0be?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) (c8aa151) will **increase** coverage by `0.08%`.
> The diff coverage is `80.88%`.

[![Impacted file tree graph](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/456/graphs/tree.svg?width=650&height=150&src=pr&token=WJK43PV390&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io)](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/456?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io)

```diff
@@            Coverage Diff             @@
##           master     #456      +/-   ##
==========================================
+ Coverage   86.19%   86.28%   +0.08%     
==========================================
  Files          33       33              
  Lines        3065     3114      +49     
==========================================
+ Hits         2642     2687      +45     
- Misses        331      332       +1     
- Partials       92       95       +3     
```


| [Impacted Files](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/456?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) | Coverage Δ | |
|---|---|---|
| [entity/rows.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/456?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L3Jvd3MuZ28=) | `90.04% <79.36%> (-1.94%)` | :arrow_down: |
| [client/meta\_cache.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/456?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L21ldGFfY2FjaGUuZ28=) | `100.00% <100.00%> (ø)` | |
| [client/row.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/456?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L3Jvdy5nbw==) | `84.54% <100.00%> (+1.05%)` | :arrow_up: |

... and [3 files with indirect coverage changes](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/456/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io)

sre-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/milvus-io/milvus-sdk-go/pull/456#" title="Author self-approved">congqixia</a>*, *<a href="https://github.com/milvus-io/milvus-sdk-go/pull/456#pullrequestreview-1435381062" title="LGTM">yah01</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=milvus-io%2Fmilvus-sdk-go).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/milvus-io/milvus-sdk-go/blob/master/OWNERS)~~ [congqixia]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/client/meta_cache.go b/client/meta_cache.go
index 530cba2..5a98457 100644
--- a/client/meta_cache.go
+++ b/client/meta_cache.go
@@ -71,6 +71,12 @@ func (m *metaCache) getCollectionInfo(cName string) (*collInfo, bool) {
 	}, true
 }
 
+func (m *metaCache) reset() {
+	m.colMu.Lock()
+	defer m.colMu.Unlock()
+	m.collInfoMap = make(map[string]collInfo)
+}
+
 func max(x, y uint64) uint64 {
 	if x > y {
 		return x
diff --git a/client/row.go b/client/row.go
index b1d3563..ec04c15 100644
--- a/client/row.go
+++ b/client/row.go
@@ -71,6 +71,9 @@ func (c *GrpcClient) InsertByRows(ctx context.Context, collName string, partitio
 	if err := c.checkCollectionExists(ctx, collName); err != nil {
 		return nil, err
 	}
+	if err := c.checkPartitionExists(ctx, collName, partitionName); err != nil {
+		return nil, err
+	}
 	coll, err := c.DescribeCollection(ctx, collName)
 	if err != nil {
 		return nil, err
@@ -80,6 +83,7 @@ func (c *GrpcClient) InsertByRows(ctx context.Context, collName string, partitio
 	if err != nil {
 		return nil, err
 	}
+	//fieldData
 	// 2. do insert request
 	req := &server.InsertRequest{
 		DbName:         "", // reserved
diff --git a/entity/rows.go b/entity/rows.go
index 8e06726..353aa83 100644
--- a/entity/rows.go
+++ b/entity/rows.go
@@ -12,6 +12,7 @@
 package entity
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"go/ast"
@@ -24,6 +25,9 @@ const (
 	// MilvusTag struct tag const for milvus row based struct
 	MilvusTag = `milvus`
 
+	// MilvusSkipTagValue struct tag const for skip this field.
+	MilvusSkipTagValue = `-`
+
 	// MilvusTagSep struct tag const for attribute separator
 	MilvusTagSep = `;`
 
@@ -107,7 +111,11 @@ func ParseSchema(r Row) (*Schema, error) {
 			ft = ft.Elem()
 		}
 		fv := reflect.New(ft)
-		tagSettings := ParseTagSetting(f.Tag.Get(MilvusTag), MilvusTagSep)
+		tag := f.Tag.Get(MilvusTag)
+		if tag == MilvusSkipTagValue {
+			continue
+		}
+		tagSettings := ParseTagSetting(tag, MilvusTagSep)
 		if _, has := tagSettings[MilvusPrimaryKey]; has {
 			field.PrimaryKey = true
 		}
@@ -229,13 +237,17 @@ func RowsToColumns(rows []Row, schemas ...*Schema) ([]Column, error) {
 	// if schema not provided, try to parse from row
 	if len(schemas) == 0 {
 		sch, err = ParseSchema(rows[0])
+		if err != nil {
+			return []Column{}, err
+		}
 	} else {
 		// use first schema provided
 		sch = schemas[0]
 	}
-	if err != nil {
-		return []Column{}, err
-	}
+
+	isDynamic := sch.EnableDynamicField
+	var dynamicCol *ColumnJSONBytes
+
 	nameColumns := make(map[string]Column)
 	for _, field := range sch.Fields {
 		switch field.DataType {
@@ -271,6 +283,14 @@ func RowsToColumns(rows []Row, schemas ...*Schema) ([]Column, error) {
 			data := make([]string, 0, rowsLen)
 			col := NewColumnString(field.Name, data)
 			nameColumns[field.Name] = col
+		case FieldTypeJSON:
+			data := make([][]byte, 0, rowsLen)
+			col := NewColumnJSONBytes(field.Name, data)
+			if isDynamic && field.IsDynamic {
+				dynamicCol = col
+			} else {
+				nameColumns[field.Name] = col
+			}
 		case FieldTypeFloatVector:
 			data := make([][]float32, 0, rowsLen)
 			dimStr, has := field.TypeParams[TypeParamDim]
@@ -297,6 +317,11 @@ func RowsToColumns(rows []Row, schemas ...*Schema) ([]Column, error) {
 			nameColumns[field.Name] = col
 		}
 	}
+
+	if isDynamic && dynamicCol == nil {
+		return nil, errors.New("schema dynamic field enabled but field not found")
+	}
+
 	for _, row := range rows {
 		// collection schema name need not to be same, since receiver could has other names
 		v := reflect.ValueOf(row)
@@ -304,43 +329,85 @@ func RowsToColumns(rows []Row, schemas ...*Schema) ([]Column, error) {
 			v = v.Elem()
 		}
 
+		type fieldCandi struct {
+			name    string
+			v       reflect.Value
+			options map[string]string
+		}
+		set := make(map[string]fieldCandi)
+		for i := 0; i < v.NumField(); i++ {
+			ft := v.Type().Field(i)
+			name := ft.Name
+			tag, ok := ft.Tag.Lookup(MilvusTag)
+
+			settings := make(map[string]string)
+			if ok {
+				if tag == MilvusSkipTagValue {
+					continue
+				}
+				settings = ParseTagSetting(tag, MilvusTagSep)
+				fn, has := settings[MilvusTagName]
+				if has {
+					// overwrite column to tag name
+					name = fn
+				}
+			}
+			_, ok = set[name]
+			// duplicated
+			if ok {
+				return nil, fmt.Errorf("column has duplicated name: %s when parsing field: %s", name, ft.Name)
+			}
+
+			v := v.Field(i)
+			if v.Kind() == reflect.Array {
+				v = v.Slice(0, v.Len()-1)
+			}
+
+			set[name] = fieldCandi{
+				name:    name,
+				v:       v,
+				options: settings,
+			}
+		}
+
 		for idx, field := range sch.Fields {
+			if isDynamic && field.IsDynamic {
+				continue
+			}
 			column := nameColumns[field.Name]
 
-			fv := fieldFromNameTag(v, field.Name)
-			if !fv.IsValid() {
-				return []Column{}, fmt.Errorf("row %d does not has field %s", idx, field.Name)
+			candi, ok := set[field.Name]
+			if !ok {
+				return nil, fmt.Errorf("row %d does not has field %s", idx, field.Name)
 			}
-			if fv.Kind() == reflect.Array { // change to slice
-				fv = fv.Slice(0, fv.Len()-1)
-			}
-			err := column.AppendValue(fv.Interface())
+			err := column.AppendValue(candi.v.Interface())
 			if err != nil {
-				return []Column{}, err
+				return nil, err
 			}
+			delete(set, field.Name)
 		}
 
+		if isDynamic {
+			m := make(map[string]interface{})
+			for name, candi := range set {
+				m[name] = candi.v.Interface()
+			}
+			bs, err := json.Marshal(m)
+			if err != nil {
+				return nil, fmt.Errorf("failed to marshal dynamic field %w", err)
+			}
+			err = dynamicCol.AppendValue(bs)
+			if err != nil {
+				return nil, fmt.Errorf("failed to append value to dynamic field %w", err)
+			}
+		}
 	}
 	columns := make([]Column, 0, len(nameColumns))
 	for _, column := range nameColumns {
 		columns = append(columns, column)
 	}
-	return columns, nil
-}
-
-func fieldFromNameTag(v reflect.Value, name string) reflect.Value {
-	// tag has higher priority
-	for i := 0; i < v.NumField(); i++ {
-		tag := v.Type().Field(i).Tag.Get(MilvusTag)
-		if tag == "" {
-			continue
-		}
-		settings := ParseTagSetting(tag, MilvusTagSep)
-		fn, has := settings[MilvusTagName]
-		if has && fn == name {
-			return v.Field(i)
-		}
+	if dynamicCol != nil {
+		columns = append(columns, dynamicCol)
 	}
-	// try use name directly
-	return v.FieldByName(name)
+	return columns, nil
 }


Test patch:
diff --git a/client/client_mock_test.go b/client/client_mock_test.go
index bcb5ee4..74834dc 100644
--- a/client/client_mock_test.go
+++ b/client/client_mock_test.go
@@ -72,6 +72,7 @@ func (s *MockSuiteBase) TearDownTest() {
 }
 
 func (s *MockSuiteBase) resetMock() {
+	MetaCache.reset()
 	if s.mock != nil {
 		s.mock.Calls = nil
 		s.mock.ExpectedCalls = nil
@@ -92,6 +93,13 @@ func (s *MockSuiteBase) setupHasCollection(collNames ...string) {
 	}, nil)
 }
 
+func (s *MockSuiteBase) setupHasCollectionError(errorCode common.ErrorCode, err error) {
+	s.mock.EXPECT().HasCollection(mock.Anything, mock.AnythingOfType("*milvuspb.HasCollectionRequest")).
+		Return(&server.BoolResponse{
+			Status: &common.Status{ErrorCode: errorCode},
+		}, err)
+}
+
 func (s *MockSuiteBase) setupHasPartition(collName string, partNames ...string) {
 	s.mock.EXPECT().HasPartition(mock.Anything, mock.AnythingOfType("*milvuspb.HasPartitionRequest")).
 		Call.Return(func(ctx context.Context, req *server.HasPartitionRequest) *server.BoolResponse {
@@ -108,6 +116,13 @@ func (s *MockSuiteBase) setupHasPartition(collName string, partNames ...string)
 	}, nil)
 }
 
+func (s *MockSuiteBase) setupHasPartitionError(errorCode common.ErrorCode, err error) {
+	s.mock.EXPECT().HasPartition(mock.Anything, mock.AnythingOfType("*milvuspb.HasPartitionRequest")).
+		Return(&server.BoolResponse{
+			Status: &common.Status{ErrorCode: errorCode},
+		}, err)
+}
+
 func (s *MockSuiteBase) setupDescribeCollection(collName string, schema *entity.Schema) {
 	s.mock.EXPECT().DescribeCollection(mock.Anything, mock.AnythingOfType("*milvuspb.DescribeCollectionRequest")).
 		Call.Return(func(ctx context.Context, req *server.DescribeCollectionRequest) *server.DescribeCollectionResponse {
@@ -118,6 +133,13 @@ func (s *MockSuiteBase) setupDescribeCollection(collName string, schema *entity.
 	}, nil)
 }
 
+func (s *MockSuiteBase) setupDescirbeCollectionError(errorCode common.ErrorCode, err error) {
+	s.mock.EXPECT().DescribeCollection(mock.Anything, mock.AnythingOfType("*milvuspb.DescribeCollectionRequest")).
+		Return(&milvuspb.DescribeCollectionResponse{
+			Status: &common.Status{ErrorCode: errorCode},
+		}, err)
+}
+
 // ref https://stackoverflow.com/questions/42102496/testing-a-grpc-service
 
 var (
diff --git a/client/client_grpc_row_test.go b/client/row_test.go
similarity index 66%
rename from client/client_grpc_row_test.go
rename to client/row_test.go
index dec1f4e..26a423a 100644
--- a/client/client_grpc_row_test.go
+++ b/client/row_test.go
@@ -14,6 +14,8 @@ import (
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 	"github.com/milvus-io/milvus-sdk-go/v2/entity"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/suite"
 )
 
 func TestCreateCollectionByRow(t *testing.T) {
@@ -86,107 +88,233 @@ func TestCreateCollectionByRow(t *testing.T) {
 	})
 }
 
-func TestInsertByRows(t *testing.T) {
-	ctx := context.Background()
+type InsertByRowsSuite struct {
+	MockSuiteBase
+}
 
-	c := testClient(ctx, t)
+func (s *InsertByRowsSuite) TestFails() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := s.client
+
+	partName := "part_1"
+
+	s.Run("fail_empty_rows", func() {
+		defer s.resetMock()
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{})
+		s.Error(err)
+	})
+
+	s.Run("fail_collection_not_found", func() {
+		defer s.resetMock()
+		s.setupHasCollection()
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
+		s.Error(err)
+	})
+
+	s.Run("fail_hascollection_errcode", func() {
+		defer s.resetMock()
+		s.setupHasCollectionError(common.ErrorCode_UnexpectedError, nil)
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
+		s.Error(err)
+	})
+
+	s.Run("fail_hascollection_error", func() {
+		defer s.resetMock()
+		s.setupHasCollectionError(common.ErrorCode_Success, errors.New("mock error"))
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
+		s.Error(err)
+	})
+
+	s.Run("fail_partition_not_found", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName)
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
+		s.Error(err)
+	})
+
+	s.Run("fail_haspartition_error", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartitionError(common.ErrorCode_Success, errors.New("mock error"))
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
+		s.Error(err)
+	})
+
+	s.Run("fail_haspartition_errcode", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartitionError(common.ErrorCode_UnexpectedError, nil)
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
+		s.Error(err)
+	})
+
+	s.Run("fail_describecollection_error", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, partName)
+		s.setupDescirbeCollectionError(common.ErrorCode_Success, errors.New("mock error"))
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
+		s.Error(err)
+	})
 
-	t.Run("test create failure due to meta", func(t *testing.T) {
-		mockServer.DelInjection(MHasCollection) // collection does not exist
-		ids, err := c.InsertByRows(ctx, testCollectionName, "", []entity.Row{})
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-
-		// partition not exists
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		ids, err = c.InsertByRows(ctx, testCollectionName, "_part_not_exists", []entity.Row{})
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-
-		// field not in collection
-		mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, 0, testCollectionName, defaultSchema()))
-		type ExtraFieldRow struct {
+	s.Run("fail_describecollection_errcode", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, partName)
+		s.setupDescirbeCollectionError(common.ErrorCode_UnexpectedError, nil)
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
+		s.Error(err)
+	})
+
+	s.Run("fail_field_missing", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, partName)
+		s.setupDescribeCollection(testCollectionName,
+			entity.NewSchema().
+				WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true).WithIsAutoID(true)).
+				WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+		type row struct {
 			entity.RowBase
-			Int64      int64 `milvus:"primary_key"`
-			ExtraField float64
-			Vector     []float32 `milvus:"dim:128"`
+			ID int64
 		}
-		ids, err = c.InsertByRows(ctx, testCollectionName, "", []entity.Row{&ExtraFieldRow{}})
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{row{ID: 100}})
+		s.Error(err)
+	})
 
-		// field type not match
-		mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, 0, testCollectionName, defaultSchema()))
-		type OtherFieldRow struct {
+	s.Run("fail_field_type_not_match", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, partName)
+		s.setupDescribeCollection(testCollectionName,
+			entity.NewSchema().
+				WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true).WithIsAutoID(true)).
+				WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+		type row struct {
 			entity.RowBase
-			Int64  int32     `milvus:"primary_key;name:int64"`
-			Vector []float32 `milvus:"dim:128;name:vector"`
+			ID      string
+			Vectors []float32
 		}
-		ids, err = c.InsertByRows(ctx, testCollectionName, "", []entity.Row{
-			&OtherFieldRow{},
-		})
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{row{ID: "100", Vectors: make([]float32, 128)}})
+		s.Error(err)
+	})
 
-		// missing field
-		type MissingFieldRow struct {
+	s.Run("fail_extra_field", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, partName)
+		s.setupDescribeCollection(testCollectionName,
+			entity.NewSchema().
+				WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true).WithIsAutoID(true)).
+				WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+		type row struct {
 			entity.RowBase
-			Vector []float32 `milvus:"dim:128;name:vector"`
+			ID      int64
+			Extra   float64
+			Vectors []float32
 		}
-		ids, err = c.InsertByRows(ctx, testCollectionName, "", []entity.Row{
-			&MissingFieldRow{},
-		})
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-		t.Log(err.Error())
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{row{ID: 100, Extra: 0.12, Vectors: make([]float32, 128)}})
+		s.Error(err)
+	})
 
-		// column len not match, not equvilant case in row based api
+	s.Run("vector_dim_not_match", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, partName)
+		s.setupDescribeCollection(testCollectionName,
+			entity.NewSchema().
+				WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true).WithIsAutoID(true)).
+				WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+		type row struct {
+			entity.RowBase
+			ID      int64
+			Vectors []float32 `milvus:"dim:16"`
+		}
+		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{row{ID: 100, Vectors: make([]float32, 16)}})
+		s.Error(err)
+	})
+}
 
-		// dim not match
-		type Dim2Row struct {
+func (s *InsertByRowsSuite) TestSuccess() {
+	partName := "part_1"
+	c := s.client
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	s.Run("non_dynamic", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, partName)
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().WithName(testCollectionName).
+			WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
+			WithField(entity.NewField().WithName("Vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		s.mock.EXPECT().Insert(mock.Anything, mock.AnythingOfType("*milvuspb.InsertRequest")).
+			Run(func(_ context.Context, req *server.InsertRequest) {
+				s.Equal(testCollectionName, req.GetCollectionName())
+				s.Equal(partName, req.GetPartitionName())
+				s.Equal(2, len(req.GetFieldsData()))
+			}).Return(&server.MutationResult{
+			Status: &common.Status{ErrorCode: common.ErrorCode_Success},
+			IDs:    &schema.IDs{IdField: &schema.IDs_IntId{IntId: &schema.LongArray{Data: []int64{100}}}},
+		}, nil)
+		type row struct {
 			entity.RowBase
-			Int64  int64     `milvus:"primary_key;name:int64"`
-			Vector []float32 `milvus:"dim:16;name:vector"`
+			ID     int64
+			Vector []float32
 		}
-		ids, err = c.InsertByRows(ctx, testCollectionName, "", []entity.Row{
-			&Dim2Row{},
+		ids, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{
+			row{ID: 100, Vector: make([]float32, 128)},
 		})
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-		t.Log(err.Error())
+		s.NoError(err)
+		s.Equal(1, ids.Len())
 	})
 
-	mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-	mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, 0, testCollectionName, defaultSchema()))
-
-	vector := generateFloatVector(4096, testVectorDim)
-	mockServer.SetInjection(MInsert, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-		req, ok := raw.(*server.InsertRequest)
-		resp := &server.MutationResult{}
-		if !ok {
-			s, err := BadRequestStatus()
-			resp.Status = s
-			return resp, err
-		}
-		assert.EqualValues(t, 4096, req.GetNumRows())
-		assert.Equal(t, testCollectionName, req.GetCollectionName())
-		intIds := &schema.IDs_IntId{
-			IntId: &schema.LongArray{
-				Data: make([]int64, 4096),
-			},
-		}
-		resp.IDs = &schema.IDs{
-			IdField: intIds,
+	s.Run("dynamic", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, partName)
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithName(testCollectionName).WithDynamicFieldEnabled(true).
+			WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
+			WithField(entity.NewField().WithName("Vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")).
+			WithField(entity.NewField().WithName("$meta").WithDataType(entity.FieldTypeJSON).WithIsDynamic(true)),
+		)
+
+		s.mock.EXPECT().Insert(mock.Anything, mock.AnythingOfType("*milvuspb.InsertRequest")).
+			Run(func(_ context.Context, req *server.InsertRequest) {
+				s.Equal(testCollectionName, req.GetCollectionName())
+				s.Equal(partName, req.GetPartitionName())
+				s.Equal(3, len(req.GetFieldsData()))
+			}).Return(&server.MutationResult{
+			Status: &common.Status{ErrorCode: common.ErrorCode_Success},
+			IDs:    &schema.IDs{IdField: &schema.IDs_IntId{IntId: &schema.LongArray{Data: []int64{100}}}},
+		}, nil)
+		type row struct {
+			entity.RowBase
+			ID     int64
+			Vector []float32
+			Extra1 int32
+			Extra2 string
 		}
-		s, err := SuccessStatus()
-		resp.Status = s
-		return resp, err
+		ids, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{
+			row{ID: 100, Vector: make([]float32, 128), Extra1: 1, Extra2: "aabb"},
+		})
+		s.NoError(err)
+		s.Equal(1, ids.Len())
 	})
-	_, err := c.Insert(ctx, testCollectionName, "", // use default partition
-		entity.NewColumnFloatVector(testVectorField, testVectorDim, vector))
-
-	assert.Nil(t, err)
+}
 
+func TestInsertByRows(t *testing.T) {
+	suite.Run(t, new(InsertByRowsSuite))
 }
 
 func TestSearchResultToRows(t *testing.T) {
diff --git a/entity/rows_test.go b/entity/rows_test.go
index a46db96..721ed4c 100644
--- a/entity/rows_test.go
+++ b/entity/rows_test.go
@@ -15,6 +15,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/suite"
 )
 
 // ArrayRow test case type
@@ -189,12 +190,15 @@ type ValidStruct struct {
 	Vector  []float32 `milvus:"dim:16"`
 	Vector2 []byte    `milvus:"dim:32"`
 }
+
 type ValidStruct2 struct {
 	RowBase
 	ID      int64 `milvus:"primary_key"`
 	Vector  [16]float32
 	Vector2 [4]byte
+	Ignored bool `milvus:"-"`
 }
+
 type ValidStructWithNamedTag struct {
 	RowBase
 	ID     int64       `milvus:"primary_key;name:id"`
@@ -235,3 +239,33 @@ func TestRowsToColumns(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 }
+
+type RowsSuite struct {
+	suite.Suite
+}
+
+func (s *RowsSuite) TestDynamicSchema() {
+	s.Run("all_fallback_dynamic", func() {
+		columns, err := RowsToColumns([]Row{&ValidStruct{}},
+			NewSchema().WithField(
+				NewField().WithName("$meta").WithDataType(FieldTypeJSON).WithIsDynamic(true),
+			).
+				WithDynamicFieldEnabled(true),
+		)
+		s.NoError(err)
+		s.Equal(1, len(columns))
+	})
+
+	s.Run("dynamic_not_found", func() {
+		_, err := RowsToColumns([]Row{&ValidStruct{}},
+			NewSchema().WithField(
+				NewField().WithName("ID").WithDataType(FieldTypeInt64).WithIsPrimaryKey(true),
+			).WithDynamicFieldEnabled(true),
+		)
+		s.Error(err)
+	})
+}
+
+func TestRows(t *testing.T) {
+	suite.Run(t, new(RowsSuite))
+}