Repository: milvus-io/milvus-sdk-go
Pull request #463 (2023-05-24 10:15:19)

Support dynamic schema for 2.2.x
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/463#" 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/v2.2.x/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/463?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) Report
> Merging [#463](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) (0ce25b3) into [v2.2.x](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/commit/a283bb5dc801b4105b0ff87627f9c59ac831eca1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) (a283bb5) will **decrease** coverage by `4.52%`.
> The diff coverage is `74.25%`.

[![Impacted file tree graph](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463/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/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io)

```diff
@@            Coverage Diff             @@
##           v2.2.x     #463      +/-   ##
==========================================
- Coverage   86.74%   82.23%   -4.52%     
==========================================
  Files          34       32       -2     
  Lines        2966     2629     -337     
==========================================
- Hits         2573     2162     -411     
- Misses        303      369      +66     
- Partials       90       98       +8     
```


| [Impacted Files](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) | Coverage Δ | |
|---|---|---|
| [entity/columns.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L2NvbHVtbnMuZ28=) | `74.28% <0.00%> (-21.31%)` | :arrow_down: |
| [entity/columns\_varchar.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L2NvbHVtbnNfdmFyY2hhci5nbw==) | `90.32% <0.00%> (-9.68%)` | :arrow_down: |
| [client/client.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L2NsaWVudC5nbw==) | `34.88% <66.66%> (+2.38%)` | :arrow_up: |
| [entity/rows.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L3Jvd3MuZ28=) | `88.15% <71.66%> (-3.83%)` | :arrow_down: |
| [client/data.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L2RhdGEuZ28=) | `69.91% <73.21%> (-2.82%)` | :arrow_down: |
| [client/insert.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L2luc2VydC5nbw==) | `78.26% <78.26%> (ø)` | |
| [entity/schema.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L3NjaGVtYS5nbw==) | `94.15% <93.44%> (+0.70%)` | :arrow_up: |
| [client/meta\_cache.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?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/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L3Jvdy5nbw==) | `84.61% <100.00%> (+1.12%)` | :arrow_up: |
| [entity/columns\_dynamic.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L2NvbHVtbnNfZHluYW1pYy5nbw==) | `100.00% <100.00%> (ø)` | |
| ... and [1 more](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/463?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) | |

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


Patch:
diff --git a/client/client.go b/client/client.go
index 4dfb2bd..c9d7635 100644
--- a/client/client.go
+++ b/client/client.go
@@ -127,9 +127,9 @@ type Client interface {
 	Search(ctx context.Context, collName string, partitions []string,
 		expr string, outputFields []string, vectors []entity.Vector, vectorField string, metricType entity.MetricType, topK int, sp entity.SearchParam, opts ...SearchQueryOptionFunc) ([]SearchResult, error)
 	// QueryByPks query record by specified primary key(s).
-	QueryByPks(ctx context.Context, collectionName string, partitionNames []string, ids entity.Column, outputFields []string, opts ...SearchQueryOptionFunc) ([]entity.Column, error)
+	QueryByPks(ctx context.Context, collectionName string, partitionNames []string, ids entity.Column, outputFields []string, opts ...SearchQueryOptionFunc) (ResultSet, error)
 	// Query performs query records with boolean expression.
-	Query(ctx context.Context, collectionName string, partitionNames []string, expr string, outputFields []string, opts ...SearchQueryOptionFunc) ([]entity.Column, error)
+	Query(ctx context.Context, collectionName string, partitionNames []string, expr string, outputFields []string, opts ...SearchQueryOptionFunc) (ResultSet, error)
 
 	// CalcDistance calculate the distance between vectors specified by ids or provided
 	CalcDistance(ctx context.Context, collName string, partitions []string,
@@ -200,11 +200,22 @@ type Client interface {
 // Fields contains the data of `outputFieleds` specified or all columns if non
 // Scores is actually the distance between the vector current record contains and the search target vector
 type SearchResult struct {
-	ResultCount int             // the returning entry count
-	IDs         entity.Column   // auto generated id, can be mapped to the columns from `Insert` API
-	Fields      []entity.Column // output field data
-	Scores      []float32       // distance to the target vector
-	Err         error           // search error if any
+	ResultCount int           // the returning entry count
+	IDs         entity.Column // auto generated id, can be mapped to the columns from `Insert` API
+	Fields      ResultSet     // output field data
+	Scores      []float32     // distance to the target vector
+	Err         error         // search error if any
+}
+
+type ResultSet []entity.Column
+
+func (rs ResultSet) GetColumn(fieldName string) entity.Column {
+	for _, column := range rs {
+		if column.Name() == fieldName {
+			return column
+		}
+	}
+	return nil
 }
 
 // Check if GrpcClient implement Client.
diff --git a/client/data.go b/client/data.go
index fa8b301..0e52dd3 100644
--- a/client/data.go
+++ b/client/data.go
@@ -14,14 +14,13 @@ package client
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"log"
 	"strconv"
 	"strings"
-	"time"
 
-	"github.com/golang/protobuf/proto"
+	"github.com/cockroachdb/errors"
+
 	common "github.com/milvus-io/milvus-proto/go-api/commonpb"
 	server "github.com/milvus-io/milvus-proto/go-api/milvuspb"
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
@@ -34,212 +33,24 @@ const (
 	ignoreGrowingKey = `ignore_growing`
 )
 
-// Insert Index  into collection with column-based format
-// collName is the collection name
-// partitionName is the partition to insert, if not specified(empty), default partition will be used
-// columns are slice of the column-based data
-func (c *GrpcClient) Insert(ctx context.Context, collName string, partitionName string, columns ...entity.Column) (entity.Column, error) {
-	if c.Service == nil {
-		return nil, ErrClientNotReady
-	}
-	// 1. validation for all input params
-	// collection
-	if err := c.checkCollectionExists(ctx, collName); err != nil {
-		return nil, err
-	}
-	if partitionName != "" {
-		err := c.checkPartitionExists(ctx, collName, partitionName)
-		if err != nil {
-			return nil, err
-		}
-	}
-	// fields
-	var rowSize int
-	coll, err := c.DescribeCollection(ctx, collName)
-	if err != nil {
-		return nil, err
-	}
-	mNameField := make(map[string]*entity.Field)
-	for _, field := range coll.Schema.Fields {
-		mNameField[field.Name] = field
-	}
-	mNameColumn := make(map[string]entity.Column)
-	for _, column := range columns {
-		mNameColumn[column.Name()] = column
-		l := column.Len()
-		if rowSize == 0 {
-			rowSize = l
-		} else {
-			if rowSize != l {
-				return nil, errors.New("column size not match")
-			}
-		}
-		field, has := mNameField[column.Name()]
-		if !has {
-			return nil, fmt.Errorf("field %s does not exist in collection %s", column.Name(), collName)
-		}
-		if column.Type() != field.DataType {
-			return nil, fmt.Errorf("param column %s has type %v but collection field definition is %v", column.Name(), column.FieldData(), field.DataType)
-		}
-		if field.DataType == entity.FieldTypeFloatVector || field.DataType == entity.FieldTypeBinaryVector {
-			dim := 0
-			switch column := column.(type) {
-			case *entity.ColumnFloatVector:
-				dim = column.Dim()
-			case *entity.ColumnBinaryVector:
-				dim = column.Dim()
-			}
-			if fmt.Sprintf("%d", dim) != field.TypeParams[entity.TypeParamDim] {
-				return nil, fmt.Errorf("params column %s vector dim %d not match collection definition, which has dim of %s", field.Name, dim, field.TypeParams[entity.TypeParamDim])
-			}
-		}
-	}
-	for _, field := range coll.Schema.Fields {
-		_, has := mNameColumn[field.Name]
-		if !has && !field.AutoID {
-			return nil, fmt.Errorf("field %s not passed", field.Name)
-		}
-	}
-
-	// 2. do insert request
-	req := &server.InsertRequest{
-		DbName:         "", // reserved
-		CollectionName: collName,
-		PartitionName:  partitionName,
-	}
-	if req.PartitionName == "" {
-		req.PartitionName = "_default" // use default partition
-	}
-	req.NumRows = uint32(rowSize)
-	for _, column := range columns {
-		req.FieldsData = append(req.FieldsData, column.FieldData())
-	}
-	resp, err := c.Service.Insert(ctx, req)
-	if err != nil {
-		return nil, err
-	}
-	if err := handleRespStatus(resp.GetStatus()); err != nil {
-		return nil, err
-	}
-	MetaCache.setSessionTs(collName, resp.Timestamp)
-	// 3. parse id column
-	return entity.IDColumns(resp.GetIDs(), 0, -1)
-}
-
-// Flush force collection to flush memory records into storage
-// in sync mode, flush will wait all segments to be flushed
-func (c *GrpcClient) Flush(ctx context.Context, collName string, async bool) error {
-	if c.Service == nil {
-		return ErrClientNotReady
-	}
-	if err := c.checkCollectionExists(ctx, collName); err != nil {
-		return err
-	}
-	req := &server.FlushRequest{
-		DbName:          "", // reserved,
-		CollectionNames: []string{collName},
-	}
-	resp, err := c.Service.Flush(ctx, req)
-	if err != nil {
-		return err
-	}
-	if err := handleRespStatus(resp.GetStatus()); err != nil {
-		return err
-	}
-	if !async {
-		segmentIDs, has := resp.GetCollSegIDs()[collName]
-		ids := segmentIDs.GetData()
-		if has && len(ids) > 0 {
-			flushed := func() bool {
-				resp, err := c.Service.GetFlushState(ctx, &server.GetFlushStateRequest{
-					SegmentIDs: ids,
-				})
-				if err != nil {
-					// TODO max retry
-					return false
-				}
-				return resp.GetFlushed()
-			}
-			for !flushed() {
-				// respect context deadline/cancel
-				select {
-				case <-ctx.Done():
-					return errors.New("deadline exceeded")
-				default:
-				}
-				time.Sleep(200 * time.Millisecond)
-			}
-		}
-	}
-	return nil
-}
-
-// DeleteByPks deletes entries related to provided primary keys
-func (c *GrpcClient) DeleteByPks(ctx context.Context, collName string, partitionName string, ids entity.Column) error {
-	if c.Service == nil {
-		return ErrClientNotReady
-	}
-
-	// check collection name
-	if err := c.checkCollectionExists(ctx, collName); err != nil {
-		return err
-	}
-	coll, err := c.DescribeCollection(ctx, collName)
-	if err != nil {
-		return err
-	}
-	// check partition name
-	if partitionName != "" {
-		err := c.checkPartitionExists(ctx, collName, partitionName)
-		if err != nil {
-			return err
-		}
-	}
-	// check primary keys
-	if ids.Len() == 0 {
-		return errors.New("ids len must not be zero")
-	}
-	if ids.Type() != entity.FieldTypeInt64 && ids.Type() != entity.FieldTypeVarChar { // string key not supported yet
-		return errors.New("only int64 and varchar column can be primary key for now")
-	}
-
-	pkf := getPKField(coll.Schema)
-	// pkf shall not be nil since is returned from milvus
-	if pkf.Name != ids.Name() {
-		return errors.New("only delete by primary key is supported now")
-	}
-
-	expr := PKs2Expr(ids)
-
-	req := &server.DeleteRequest{
-		DbName:         "",
-		CollectionName: collName,
-		PartitionName:  partitionName,
-		Expr:           expr,
-	}
-
-	resp, err := c.Service.Delete(ctx, req)
-	if err != nil {
-		return err
-	}
-	err = handleRespStatus(resp.GetStatus())
-	if err != nil {
-		return err
-	}
-	MetaCache.setSessionTs(collName, resp.Timestamp)
-	return nil
-}
-
 // Search with bool expression
 func (c *GrpcClient) Search(ctx context.Context, collName string, partitions []string,
 	expr string, outputFields []string, vectors []entity.Vector, vectorField string, metricType entity.MetricType, topK int, sp entity.SearchParam, opts ...SearchQueryOptionFunc) ([]SearchResult, error) {
 	if c.Service == nil {
 		return []SearchResult{}, ErrClientNotReady
 	}
-	_, ok := MetaCache.getCollectionInfo(collName)
+	var schema *entity.Schema
+	collInfo, ok := MetaCache.getCollectionInfo(collName)
 	if !ok {
-		c.DescribeCollection(ctx, collName)
+		coll, err := c.DescribeCollection(ctx, collName)
+		if err != nil {
+			return nil, err
+		}
+		schema = coll.Schema
+	} else {
+		schema = collInfo.Schema
 	}
+
 	option, err := makeSearchQueryOption(collName, opts...)
 	if err != nil {
 		return nil, err
@@ -273,38 +84,73 @@ func (c *GrpcClient) Search(ctx context.Context, collName string, partitions []s
 			offset += rc
 			continue
 		}
-		entry.Fields = make([]entity.Column, 0, len(fieldDataList))
-		for _, fieldData := range fieldDataList {
-			column, err := entity.FieldDataColumn(fieldData, offset, offset+rc)
-			if err != nil {
-				entry.Err = err
-				continue
-			}
-			entry.Fields = append(entry.Fields, column)
-		}
+		entry.Fields, entry.Err = c.parseSearchResult(schema, outputFields, fieldDataList, i, offset, offset+rc)
 		sr = append(sr, entry)
 		offset += rc
 	}
 	return sr, nil
 }
 
-func PKs2Expr(ids entity.Column) string {
+func (c *GrpcClient) parseSearchResult(sch *entity.Schema, outputFields []string, fieldDataList []*schema.FieldData, idx, from, to int) ([]entity.Column, error) {
+	//dynamicField := sch.GetDynamicField()
+	fields := make(map[string]*schema.FieldData)
+	var dynamicColumn *entity.ColumnJSONBytes
+	for _, fieldData := range fieldDataList {
+		fields[fieldData.GetFieldName()] = fieldData
+		if fieldData.GetIsDynamic() {
+			column, err := entity.FieldDataColumn(fieldData, from, to)
+			if err != nil {
+				return nil, err
+			}
+			var ok bool
+			dynamicColumn, ok = column.(*entity.ColumnJSONBytes)
+			if !ok {
+				return nil, errors.New("dynamic field not json")
+			}
+		}
+	}
+	columns := make([]entity.Column, 0, len(outputFields))
+	for _, outputField := range outputFields {
+		fieldData, ok := fields[outputField]
+		var column entity.Column
+		var err error
+		if !ok {
+			if dynamicColumn == nil {
+				return nil, errors.New("output fields not match and result field data does not contain dynamic field")
+			}
+			column = entity.NewColumnDynamic(dynamicColumn, outputField)
+		} else {
+			column, err = entity.FieldDataColumn(fieldData, from, to)
+		}
+		if err != nil {
+			return nil, err
+		}
+		columns = append(columns, column)
+	}
+	return columns, nil
+}
+
+func PKs2Expr(backName string, ids entity.Column) string {
 	var expr string
+	var pkName = ids.Name()
+	if ids.Name() == "" {
+		pkName = backName
+	}
 	switch ids.Type() {
 	case entity.FieldTypeInt64:
-		expr = fmt.Sprintf("%s in %s", ids.Name(), strings.Join(strings.Fields(fmt.Sprint(ids.FieldData().GetScalars().GetLongData().GetData())), ","))
+		expr = fmt.Sprintf("%s in %s", pkName, strings.Join(strings.Fields(fmt.Sprint(ids.FieldData().GetScalars().GetLongData().GetData())), ","))
 	case entity.FieldTypeVarChar:
 		data := ids.FieldData().GetScalars().GetData().(*schema.ScalarField_StringData).StringData.Data
 		for i := range data {
 			data[i] = fmt.Sprintf("\"%s\"", data[i])
 		}
-		expr = fmt.Sprintf("%s in %s", ids.Name(), strings.Join(strings.Fields(fmt.Sprint(data)), ","))
+		expr = fmt.Sprintf("%s in %s", pkName, strings.Join(strings.Fields(fmt.Sprint(data)), ","))
 	}
 	return expr
 }
 
 // QueryByPks query record by specified primary key(s)
-func (c *GrpcClient) QueryByPks(ctx context.Context, collectionName string, partitionNames []string, ids entity.Column, outputFields []string, opts ...SearchQueryOptionFunc) ([]entity.Column, error) {
+func (c *GrpcClient) QueryByPks(ctx context.Context, collectionName string, partitionNames []string, ids entity.Column, outputFields []string, opts ...SearchQueryOptionFunc) (ResultSet, error) {
 	if c.Service == nil {
 		return nil, ErrClientNotReady
 	}
@@ -316,23 +162,27 @@ func (c *GrpcClient) QueryByPks(ctx context.Context, collectionName string, part
 		return nil, errors.New("only int64 and varchar column can be primary key for now")
 	}
 
-	expr := PKs2Expr(ids)
+	expr := PKs2Expr("", ids)
 
 	return c.Query(ctx, collectionName, partitionNames, expr, outputFields, opts...)
 }
 
 // Query performs query by expression.
-func (c *GrpcClient) Query(ctx context.Context, collectionName string, partitionNames []string, expr string, outputFields []string, opts ...SearchQueryOptionFunc) ([]entity.Column, error) {
+func (c *GrpcClient) Query(ctx context.Context, collectionName string, partitionNames []string, expr string, outputFields []string, opts ...SearchQueryOptionFunc) (ResultSet, error) {
 	if c.Service == nil {
 		return nil, ErrClientNotReady
 	}
 
-	_, ok := MetaCache.getCollectionInfo(collectionName)
+	var sch *entity.Schema
+	collInfo, ok := MetaCache.getCollectionInfo(collectionName)
 	if !ok {
-		_, err := c.DescribeCollection(ctx, collectionName)
+		coll, err := c.DescribeCollection(ctx, collectionName)
 		if err != nil {
 			return nil, err
 		}
+		sch = coll.Schema
+	} else {
+		sch = collInfo.Schema
 	}
 
 	option, err := makeSearchQueryOption(collectionName, opts...)
@@ -369,22 +219,22 @@ func (c *GrpcClient) Query(ctx context.Context, collectionName string, partition
 	}
 
 	fieldsData := resp.GetFieldsData()
-	columns := make([]entity.Column, 0, len(fieldsData))
-	for _, fieldData := range resp.GetFieldsData() {
-		if fieldData.GetType() == schema.DataType_FloatVector ||
-			fieldData.GetType() == schema.DataType_BinaryVector {
-			column, err := entity.FieldDataVector(fieldData)
-			if err != nil {
-				return nil, err
-			}
-			columns = append(columns, column)
-			continue
-		}
-		column, err := entity.FieldDataColumn(fieldData, 0, -1)
-		if err != nil {
-			return nil, err
+	// query always has pk field as output
+	hasPK := false
+	pkName := sch.PKFieldName()
+	for _, output := range outputFields {
+		if output == pkName {
+			hasPK = true
+			break
 		}
-		columns = append(columns, column)
+	}
+	if !hasPK {
+		outputFields = append(outputFields, pkName)
+	}
+
+	columns, err := c.parseSearchResult(sch, outputFields, fieldsData, 0, 0, -1) //entity.FieldDataColumn(fieldData, 0, -1)
+	if err != nil {
+		return nil, err
 	}
 
 	return columns, nil
@@ -627,39 +477,6 @@ func columnToVectorsArray(collName string, partitions []string, column entity.Co
 	return result
 }
 
-func vector2PlaceholderGroupBytes(vectors []entity.Vector) []byte {
-	phg := &common.PlaceholderGroup{
-		Placeholders: []*common.PlaceholderValue{
-			vector2Placeholder(vectors),
-		},
-	}
-
-	bs, _ := proto.Marshal(phg)
-	return bs
-}
-
-func vector2Placeholder(vectors []entity.Vector) *common.PlaceholderValue {
-	var placeHolderType common.PlaceholderType
-	ph := &common.PlaceholderValue{
-		Tag:    "$0",
-		Values: make([][]byte, 0, len(vectors)),
-	}
-	if len(vectors) == 0 {
-		return ph
-	}
-	switch vectors[0].(type) {
-	case entity.FloatVector:
-		placeHolderType = common.PlaceholderType_FloatVector
-	case entity.BinaryVector:
-		placeHolderType = common.PlaceholderType_BinaryVector
-	}
-	ph.Type = placeHolderType
-	for _, vector := range vectors {
-		ph.Values = append(ph.Values, vector.Serialize())
-	}
-	return ph
-}
-
 func isCollectionPrimaryKey(coll *entity.Collection, column entity.Column) bool {
 	if coll == nil || coll.Schema == nil || column == nil {
 		return false
@@ -728,91 +545,3 @@ func estRowSize(sch *entity.Schema, selected []string) int64 {
 	}
 	return total
 }
-
-// BulkInsert data files(json, numpy, etc.) on MinIO/S3 storage, read and parse them into sealed segments
-func (c *GrpcClient) BulkInsert(ctx context.Context, collName string, partitionName string, files []string, opts ...BulkInsertOption) (int64, error) {
-	if c.Service == nil {
-		return 0, ErrClientNotReady
-	}
-	req := &server.ImportRequest{
-		CollectionName: collName,
-		PartitionName:  partitionName,
-		Files:          files,
-	}
-
-	for _, opt := range opts {
-		opt(req)
-	}
-
-	resp, err := c.Service.Import(ctx, req)
-	if err != nil {
-		return 0, err
-	}
-	if err := handleRespStatus(resp.GetStatus()); err != nil {
-		return 0, err
-	}
-
-	return resp.Tasks[0], nil
-}
-
-// GetBulkInsertState checks import task state
-func (c *GrpcClient) GetBulkInsertState(ctx context.Context, taskID int64) (*entity.BulkInsertTaskState, error) {
-	if c.Service == nil {
-		return nil, ErrClientNotReady
-	}
-	req := &server.GetImportStateRequest{
-		Task: taskID,
-	}
-	resp, err := c.Service.GetImportState(ctx, req)
-	if err != nil {
-		return nil, err
-	}
-	if err := handleRespStatus(resp.GetStatus()); err != nil {
-		return nil, err
-	}
-
-	return &entity.BulkInsertTaskState{
-		ID:           resp.GetId(),
-		State:        entity.BulkInsertState(resp.GetState()),
-		RowCount:     resp.GetRowCount(),
-		IDList:       resp.GetIdList(),
-		Infos:        entity.KvPairsMap(resp.GetInfos()),
-		CollectionID: resp.GetCollectionId(),
-		SegmentIDs:   resp.GetSegmentIds(),
-		CreateTs:     resp.GetCreateTs(),
-	}, nil
-}
-
-// ListBulkInsertTasks list state of all import tasks
-func (c *GrpcClient) ListBulkInsertTasks(ctx context.Context, collName string, limit int64) ([]*entity.BulkInsertTaskState, error) {
-	if c.Service == nil {
-		return nil, ErrClientNotReady
-	}
-	req := &server.ListImportTasksRequest{
-		CollectionName: collName,
-		Limit:          limit,
-	}
-	resp, err := c.Service.ListImportTasks(ctx, req)
-	if err != nil {
-		return nil, err
-	}
-	if err := handleRespStatus(resp.GetStatus()); err != nil {
-		return nil, err
-	}
-
-	tasks := make([]*entity.BulkInsertTaskState, 0)
-	for _, task := range resp.GetTasks() {
-		tasks = append(tasks, &entity.BulkInsertTaskState{
-			ID:           task.GetId(),
-			State:        entity.BulkInsertState(task.GetState()),
-			RowCount:     task.GetRowCount(),
-			IDList:       task.GetIdList(),
-			Infos:        entity.KvPairsMap(task.GetInfos()),
-			CollectionID: task.GetCollectionId(),
-			SegmentIDs:   task.GetSegmentIds(),
-			CreateTs:     task.GetCreateTs(),
-		})
-	}
-
-	return tasks, nil
-}
diff --git a/client/insert.go b/client/insert.go
new file mode 100644
index 0000000..e610098
--- /dev/null
+++ b/client/insert.go
@@ -0,0 +1,416 @@
+// Copyright (C) 2019-2021 Zilliz. All rights reserved.
+//
+// Licensed 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 client
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/cockroachdb/errors"
+	"github.com/golang/protobuf/proto"
+	common "github.com/milvus-io/milvus-proto/go-api/commonpb"
+	server "github.com/milvus-io/milvus-proto/go-api/milvuspb"
+	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
+
+	"github.com/milvus-io/milvus-sdk-go/v2/entity"
+)
+
+// Insert Index  into collection with column-based format
+// collName is the collection name
+// partitionName is the partition to insert, if not specified(empty), default partition will be used
+// columns are slice of the column-based data
+func (c *GrpcClient) Insert(ctx context.Context, collName string, partitionName string, columns ...entity.Column) (entity.Column, error) {
+	if c.Service == nil {
+		return nil, ErrClientNotReady
+	}
+	// 1. validation for all input params
+	// collection
+	if err := c.checkCollectionExists(ctx, collName); err != nil {
+		return nil, err
+	}
+	if partitionName != "" {
+		err := c.checkPartitionExists(ctx, collName, partitionName)
+		if err != nil {
+			return nil, err
+		}
+	}
+	// fields
+	var rowSize int
+	coll, err := c.DescribeCollection(ctx, collName)
+	if err != nil {
+		return nil, err
+	}
+
+	// convert columns to field data
+	fieldsData, rowSize, err := c.processInsertColumns(coll.Schema, columns...)
+	if err != nil {
+		return nil, err
+	}
+
+	// 2. do insert request
+	req := &server.InsertRequest{
+		DbName:         "", // reserved
+		CollectionName: collName,
+		PartitionName:  partitionName,
+		FieldsData:     fieldsData,
+	}
+	if req.PartitionName == "" {
+		req.PartitionName = "_default" // use default partition
+	}
+	req.NumRows = uint32(rowSize)
+
+	resp, err := c.Service.Insert(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	if err := handleRespStatus(resp.GetStatus()); err != nil {
+		return nil, err
+	}
+	MetaCache.setSessionTs(collName, resp.Timestamp)
+	// 3. parse id column
+	return entity.IDColumns(resp.GetIDs(), 0, -1)
+}
+
+func (c *GrpcClient) processInsertColumns(colSchema *entity.Schema, columns ...entity.Column) ([]*schema.FieldData, int, error) {
+	// setup dynamic related var
+	isDynamic := colSchema.EnableDynamicField
+
+	// check columns and field matches
+	var rowSize int
+	mNameField := make(map[string]*entity.Field)
+	for _, field := range colSchema.Fields {
+		mNameField[field.Name] = field
+	}
+	mNameColumn := make(map[string]entity.Column)
+	var dynamicColumns []entity.Column
+	for _, column := range columns {
+		_, dup := mNameColumn[column.Name()]
+		if dup {
+			return nil, 0, fmt.Errorf("duplicated column %s found", column.Name())
+		}
+		l := column.Len()
+		if rowSize == 0 {
+			rowSize = l
+		} else {
+			if rowSize != l {
+				return nil, 0, errors.New("column size not match")
+			}
+		}
+		field, has := mNameField[column.Name()]
+		if !has {
+			if !isDynamic {
+				return nil, 0, fmt.Errorf("field %s does not exist in collection %s", column.Name(), colSchema.CollectionName)
+			}
+			// add to dynamic column list for further processing
+			dynamicColumns = append(dynamicColumns, column)
+			continue
+		}
+
+		mNameColumn[column.Name()] = column
+		if column.Type() != field.DataType {
+			return nil, 0, fmt.Errorf("param column %s has type %v but collection field definition is %v", column.Name(), column.FieldData(), field.DataType)
+		}
+		if field.DataType == entity.FieldTypeFloatVector || field.DataType == entity.FieldTypeBinaryVector {
+			dim := 0
+			switch column := column.(type) {
+			case *entity.ColumnFloatVector:
+				dim = column.Dim()
+			case *entity.ColumnBinaryVector:
+				dim = column.Dim()
+			}
+			if fmt.Sprintf("%d", dim) != field.TypeParams[entity.TypeParamDim] {
+				return nil, 0, fmt.Errorf("params column %s vector dim %d not match collection definition, which has dim of %s", field.Name, dim, field.TypeParams[entity.TypeParamDim])
+			}
+		}
+	}
+
+	// check all fixed field pass value
+	for _, field := range colSchema.Fields {
+		_, has := mNameColumn[field.Name]
+		if !has &&
+			!field.AutoID && !field.IsDynamic {
+			return nil, 0, fmt.Errorf("field %s not passed", field.Name)
+		}
+	}
+
+	fieldsData := make([]*schema.FieldData, 0, len(mNameColumn)+1)
+	for _, fixedColumn := range mNameColumn {
+		fieldsData = append(fieldsData, fixedColumn.FieldData())
+	}
+	if len(dynamicColumns) > 0 {
+		// use empty column name here
+		col, err := c.mergeDynamicColumns("", rowSize, dynamicColumns)
+		if err != nil {
+			return nil, 0, err
+		}
+		fieldsData = append(fieldsData, col)
+	}
+
+	return fieldsData, rowSize, nil
+}
+
+func (c *GrpcClient) mergeDynamicColumns(dynamicName string, rowSize int, columns []entity.Column) (*schema.FieldData, error) {
+	values := make([][]byte, 0, rowSize)
+	for i := 0; i < rowSize; i++ {
+		m := make(map[string]interface{})
+		for _, column := range columns {
+			// range guaranteed
+			m[column.Name()], _ = column.Get(i)
+		}
+		bs, err := json.Marshal(m)
+		if err != nil {
+			return nil, err
+		}
+		values = append(values, bs)
+	}
+	return &schema.FieldData{
+		Type:      schema.DataType_JSON,
+		FieldName: dynamicName,
+		Field: &schema.FieldData_Scalars{
+			Scalars: &schema.ScalarField{
+				Data: &schema.ScalarField_JsonData{
+					JsonData: &schema.JSONArray{
+						Data: values,
+					},
+				},
+			},
+		},
+		IsDynamic: true,
+	}, nil
+}
+
+// Flush force collection to flush memory records into storage
+// in sync mode, flush will wait all segments to be flushed
+func (c *GrpcClient) Flush(ctx context.Context, collName string, async bool) error {
+	if c.Service == nil {
+		return ErrClientNotReady
+	}
+	if err := c.checkCollectionExists(ctx, collName); err != nil {
+		return err
+	}
+	req := &server.FlushRequest{
+		DbName:          "", // reserved,
+		CollectionNames: []string{collName},
+	}
+	resp, err := c.Service.Flush(ctx, req)
+	if err != nil {
+		return err
+	}
+	if err := handleRespStatus(resp.GetStatus()); err != nil {
+		return err
+	}
+	if !async {
+		segmentIDs, has := resp.GetCollSegIDs()[collName]
+		ids := segmentIDs.GetData()
+		if has && len(ids) > 0 {
+			flushed := func() bool {
+				resp, err := c.Service.GetFlushState(ctx, &server.GetFlushStateRequest{
+					SegmentIDs: ids,
+				})
+				if err != nil {
+					// TODO max retry
+					return false
+				}
+				return resp.GetFlushed()
+			}
+			for !flushed() {
+				// respect context deadline/cancel
+				select {
+				case <-ctx.Done():
+					return errors.New("deadline exceeded")
+				default:
+				}
+				time.Sleep(200 * time.Millisecond)
+			}
+		}
+	}
+	return nil
+}
+
+// DeleteByPks deletes entries related to provided primary keys
+func (c *GrpcClient) DeleteByPks(ctx context.Context, collName string, partitionName string, ids entity.Column) error {
+	if c.Service == nil {
+		return ErrClientNotReady
+	}
+
+	// check collection name
+	if err := c.checkCollectionExists(ctx, collName); err != nil {
+		return err
+	}
+	coll, err := c.DescribeCollection(ctx, collName)
+	if err != nil {
+		return err
+	}
+	// check partition name
+	if partitionName != "" {
+		err := c.checkPartitionExists(ctx, collName, partitionName)
+		if err != nil {
+			return err
+		}
+	}
+	// check primary keys
+	if ids.Len() == 0 {
+		return errors.New("ids len must not be zero")
+	}
+	if ids.Type() != entity.FieldTypeInt64 && ids.Type() != entity.FieldTypeVarChar { // string key not supported yet
+		return errors.New("only int64 and varchar column can be primary key for now")
+	}
+
+	pkf := getPKField(coll.Schema)
+	// pkf shall not be nil since is returned from milvus
+	if ids.Name() != "" && pkf.Name != ids.Name() {
+		return errors.New("only delete by primary key is supported now")
+	}
+
+	expr := PKs2Expr(pkf.Name, ids)
+
+	req := &server.DeleteRequest{
+		DbName:         "",
+		CollectionName: collName,
+		PartitionName:  partitionName,
+		Expr:           expr,
+	}
+
+	resp, err := c.Service.Delete(ctx, req)
+	if err != nil {
+		return err
+	}
+	err = handleRespStatus(resp.GetStatus())
+	if err != nil {
+		return err
+	}
+	MetaCache.setSessionTs(collName, resp.Timestamp)
+	return nil
+}
+
+// BulkInsert data files(json, numpy, etc.) on MinIO/S3 storage, read and parse them into sealed segments
+func (c *GrpcClient) BulkInsert(ctx context.Context, collName string, partitionName string, files []string, opts ...BulkInsertOption) (int64, error) {
+	if c.Service == nil {
+		return 0, ErrClientNotReady
+	}
+	req := &server.ImportRequest{
+		CollectionName: collName,
+		PartitionName:  partitionName,
+		Files:          files,
+	}
+
+	for _, opt := range opts {
+		opt(req)
+	}
+
+	resp, err := c.Service.Import(ctx, req)
+	if err != nil {
+		return 0, err
+	}
+	if err := handleRespStatus(resp.GetStatus()); err != nil {
+		return 0, err
+	}
+
+	return resp.Tasks[0], nil
+}
+
+// GetBulkInsertState checks import task state
+func (c *GrpcClient) GetBulkInsertState(ctx context.Context, taskID int64) (*entity.BulkInsertTaskState, error) {
+	if c.Service == nil {
+		return nil, ErrClientNotReady
+	}
+	req := &server.GetImportStateRequest{
+		Task: taskID,
+	}
+	resp, err := c.Service.GetImportState(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	if err := handleRespStatus(resp.GetStatus()); err != nil {
+		return nil, err
+	}
+
+	return &entity.BulkInsertTaskState{
+		ID:           resp.GetId(),
+		State:        entity.BulkInsertState(resp.GetState()),
+		RowCount:     resp.GetRowCount(),
+		IDList:       resp.GetIdList(),
+		Infos:        entity.KvPairsMap(resp.GetInfos()),
+		CollectionID: resp.GetCollectionId(),
+		SegmentIDs:   resp.GetSegmentIds(),
+		CreateTs:     resp.GetCreateTs(),
+	}, nil
+}
+
+// ListBulkInsertTasks list state of all import tasks
+func (c *GrpcClient) ListBulkInsertTasks(ctx context.Context, collName string, limit int64) ([]*entity.BulkInsertTaskState, error) {
+	if c.Service == nil {
+		return nil, ErrClientNotReady
+	}
+	req := &server.ListImportTasksRequest{
+		CollectionName: collName,
+		Limit:          limit,
+	}
+	resp, err := c.Service.ListImportTasks(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	if err := handleRespStatus(resp.GetStatus()); err != nil {
+		return nil, err
+	}
+
+	tasks := make([]*entity.BulkInsertTaskState, 0)
+	for _, task := range resp.GetTasks() {
+		tasks = append(tasks, &entity.BulkInsertTaskState{
+			ID:           task.GetId(),
+			State:        entity.BulkInsertState(task.GetState()),
+			RowCount:     task.GetRowCount(),
+			IDList:       task.GetIdList(),
+			Infos:        entity.KvPairsMap(task.GetInfos()),
+			CollectionID: task.GetCollectionId(),
+			SegmentIDs:   task.GetSegmentIds(),
+			CreateTs:     task.GetCreateTs(),
+		})
+	}
+
+	return tasks, nil
+}
+
+func vector2PlaceholderGroupBytes(vectors []entity.Vector) []byte {
+	phg := &common.PlaceholderGroup{
+		Placeholders: []*common.PlaceholderValue{
+			vector2Placeholder(vectors),
+		},
+	}
+
+	bs, _ := proto.Marshal(phg)
+	return bs
+}
+
+func vector2Placeholder(vectors []entity.Vector) *common.PlaceholderValue {
+	var placeHolderType common.PlaceholderType
+	ph := &common.PlaceholderValue{
+		Tag:    "$0",
+		Values: make([][]byte, 0, len(vectors)),
+	}
+	if len(vectors) == 0 {
+		return ph
+	}
+	switch vectors[0].(type) {
+	case entity.FloatVector:
+		placeHolderType = common.PlaceholderType_FloatVector
+	case entity.BinaryVector:
+		placeHolderType = common.PlaceholderType_BinaryVector
+	}
+	ph.Type = placeHolderType
+	for _, vector := range vectors {
+		ph.Values = append(ph.Values, vector.Serialize())
+	}
+	return ph
+}
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 99407f3..6145028 100644
--- a/client/row.go
+++ b/client/row.go
@@ -2,10 +2,11 @@ package client
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"reflect"
 
+	"github.com/cockroachdb/errors"
+
 	"github.com/golang/protobuf/proto"
 	server "github.com/milvus-io/milvus-proto/go-api/milvuspb"
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
@@ -70,6 +71,11 @@ func (c *GrpcClient) InsertByRows(ctx context.Context, collName string, partitio
 	if err := c.checkCollectionExists(ctx, collName); err != nil {
 		return nil, err
 	}
+	if partitionName != "" {
+		if err := c.checkPartitionExists(ctx, collName, partitionName); err != nil {
+			return nil, err
+		}
+	}
 	coll, err := c.DescribeCollection(ctx, collName)
 	if err != nil {
 		return nil, err
@@ -79,12 +85,14 @@ 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
 		CollectionName: collName,
 		PartitionName:  partitionName,
 	}
+	//c.Service.
 	if req.PartitionName == "" {
 		req.PartitionName = "_default" // use default partition
 	}
diff --git a/codecov.yml b/codecov.yml
index 9533f3b..395a575 100644
--- a/codecov.yml
+++ b/codecov.yml
@@ -37,4 +37,6 @@ ignore:
   - "docs/.*"
   - "**/*.pb.go"
   - "examples/*"
-  - "tests/*"
+  - "test/*"
+  - "mocks/*"
+  - "**/*_gen.go"
diff --git a/entity/columns.go b/entity/columns.go
index d9ea388..8e5a38a 100644
--- a/entity/columns.go
+++ b/entity/columns.go
@@ -13,10 +13,10 @@ package entity
 
 import (
 	"encoding/binary"
-	"errors"
 	"fmt"
 	"math"
 
+	"github.com/cockroachdb/errors"
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 )
 
@@ -29,6 +29,7 @@ type Column interface {
 	Len() int
 	FieldData() *schema.FieldData
 	AppendValue(interface{}) error
+	Get(int) (interface{}, error)
 }
 
 // Vector interface vector used int search
@@ -213,6 +214,60 @@ func FieldDataColumn(fd *schema.FieldData, begin, end int) (Column, error) {
 			return NewColumnVarChar(fd.GetFieldName(), data.StringData.GetData()[begin:]), nil
 		}
 		return NewColumnVarChar(fd.GetFieldName(), data.StringData.GetData()[begin:end]), nil
+
+	case schema.DataType_JSON:
+		data, ok := fd.GetScalars().GetData().(*schema.ScalarField_JsonData)
+		isDynamic := fd.GetIsDynamic()
+		if !ok {
+			return nil, errFieldDataTypeNotMatch
+		}
+		if end < 0 {
+			return NewColumnJSONBytes(fd.GetFieldName(), data.JsonData.GetData()[begin:]).WithIsDynamic(isDynamic), nil
+		}
+		return NewColumnJSONBytes(fd.GetFieldName(), data.JsonData.GetData()[begin:end]).WithIsDynamic(isDynamic), nil
+
+	case schema.DataType_FloatVector:
+		vectors := fd.GetVectors()
+		x, ok := vectors.GetData().(*schema.VectorField_FloatVector)
+		if !ok {
+			return nil, errFieldDataTypeNotMatch
+		}
+		data := x.FloatVector.GetData()
+		dim := int(vectors.GetDim())
+		if end < 0 {
+			end = int(len(data) / dim)
+		}
+		vector := make([][]float32, 0, end-begin) // shall not have remanunt
+		for i := begin; i < end; i++ {
+			v := make([]float32, dim)
+			copy(v, data[i*dim:(i+1)*dim])
+			vector = append(vector, v)
+		}
+		return NewColumnFloatVector(fd.GetFieldName(), dim, vector), nil
+
+	case schema.DataType_BinaryVector:
+		vectors := fd.GetVectors()
+		x, ok := vectors.GetData().(*schema.VectorField_BinaryVector)
+		if !ok {
+			return nil, errFieldDataTypeNotMatch
+		}
+		data := x.BinaryVector
+		if data == nil {
+			return nil, errFieldDataTypeNotMatch
+		}
+		dim := int(vectors.GetDim())
+		blen := dim / 8
+		if end < 0 {
+			end = int(len(data) / blen)
+		}
+		vector := make([][]byte, 0, end-begin)
+		for i := begin; i < end; i++ {
+			v := make([]byte, blen)
+			copy(v, data[i*blen:(i+1)*blen])
+			vector = append(vector, v)
+		}
+		return NewColumnBinaryVector(fd.GetFieldName(), dim, vector), nil
+
 	default:
 		return nil, fmt.Errorf("unsupported data type %s", fd.GetType())
 	}
diff --git a/entity/columns_dynamic.go b/entity/columns_dynamic.go
new file mode 100644
index 0000000..e4a0183
--- /dev/null
+++ b/entity/columns_dynamic.go
@@ -0,0 +1,94 @@
+// Copyright (C) 2019-2021 Zilliz. All rights reserved.
+//
+// Licensed 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 entity
+
+import (
+	"github.com/cockroachdb/errors"
+	"github.com/tidwall/gjson"
+)
+
+// ColumnDynamic is a logically wrapper for dynamic json field with provided output field.
+type ColumnDynamic struct {
+	*ColumnJSONBytes
+	outputField string
+}
+
+func NewColumnDynamic(column *ColumnJSONBytes, outputField string) *ColumnDynamic {
+	return &ColumnDynamic{
+		ColumnJSONBytes: column,
+		outputField:     outputField,
+	}
+}
+
+func (c *ColumnDynamic) Name() string {
+	return c.outputField
+}
+
+func (c *ColumnDynamic) GetInt64(idx int) (int64, error) {
+	bs, err := c.ColumnJSONBytes.ValueByIdx(idx)
+	if err != nil {
+		return 0, err
+	}
+	r := gjson.GetBytes(bs, c.outputField)
+	if !r.Exists() {
+		return 0, errors.New("column not has value")
+	}
+	if r.Type != gjson.Number {
+		return 0, errors.New("column not int")
+	}
+	return r.Int(), nil
+}
+
+func (c *ColumnDynamic) GetString(idx int) (string, error) {
+	bs, err := c.ColumnJSONBytes.ValueByIdx(idx)
+	if err != nil {
+		return "", err
+	}
+	r := gjson.GetBytes(bs, c.outputField)
+	if !r.Exists() {
+		return "", errors.New("column not has value")
+	}
+	if r.Type != gjson.String {
+		return "", errors.New("column not string")
+	}
+	return r.String(), nil
+}
+
+func (c *ColumnDynamic) GetBool(idx int) (bool, error) {
+	bs, err := c.ColumnJSONBytes.ValueByIdx(idx)
+	if err != nil {
+		return false, err
+	}
+	r := gjson.GetBytes(bs, c.outputField)
+	if !r.Exists() {
+		return false, errors.New("column not has value")
+	}
+	if !r.IsBool() {
+		return false, errors.New("column not string")
+	}
+	return r.Bool(), nil
+}
+
+func (c *ColumnDynamic) GetDouble(idx int) (float64, error) {
+	bs, err := c.ColumnJSONBytes.ValueByIdx(idx)
+	if err != nil {
+		return 0, err
+	}
+	r := gjson.GetBytes(bs, c.outputField)
+	if !r.Exists() {
+		return 0, errors.New("column not has value")
+	}
+	if r.Type != gjson.Number {
+		return 0, errors.New("column not string")
+	}
+	return r.Float(), nil
+}
diff --git a/entity/columns_json.go b/entity/columns_json.go
index 5f3f9f2..41c7919 100644
--- a/entity/columns_json.go
+++ b/entity/columns_json.go
@@ -12,8 +12,9 @@ var _ (Column) = (*ColumnJSONBytes)(nil)
 // ColumnJSONBytes column type for JSON.
 // all items are marshaled json bytes.
 type ColumnJSONBytes struct {
-	name   string
-	values [][]byte
+	name      string
+	values    [][]byte
+	isDynamic bool
 }
 
 // Name returns column name.
@@ -44,6 +45,7 @@ func (c *ColumnJSONBytes) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
 		Type:      schema.DataType_JSON,
 		FieldName: c.name,
+		IsDynamic: c.isDynamic,
 	}
 
 	fd.Field = &schema.FieldData_Scalars{
@@ -83,6 +85,11 @@ func (c *ColumnJSONBytes) Data() [][]byte {
 	return c.values
 }
 
+func (c *ColumnJSONBytes) WithIsDynamic(isDynamic bool) *ColumnJSONBytes {
+	c.isDynamic = isDynamic
+	return c
+}
+
 // NewColumnJSONBytes composes a Column with json bytes.
 func NewColumnJSONBytes(name string, values [][]byte) *ColumnJSONBytes {
 	return &ColumnJSONBytes{
diff --git a/entity/columns_scalar_gen.go b/entity/columns_scalar_gen.go
index 55b229a..14869e8 100755
--- a/entity/columns_scalar_gen.go
+++ b/entity/columns_scalar_gen.go
@@ -1,12 +1,12 @@
 // Code generated by go generate; DO NOT EDIT
-// This file is generated by go generate at 2021-07-28 17:22:17.531936577 +0800 CST m=+0.003667571
+// This file is generated by go generate
 
-package entity
+package entity 
 
 import (
-	"errors"
 	"fmt"
 
+	"github.com/cockroachdb/errors"
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 )
 
@@ -31,14 +31,23 @@ func (c *ColumnBool) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnBool) Get(idx int) (interface{}, error) {
+	var r bool // use default value
+	if idx < 0 || idx >= c.Len() {
+		return r, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnBool) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Bool,
+		Type: schema.DataType_Bool,
 		FieldName: c.name,
 	}
 	data := make([]bool, 0, c.Len())
-	for i := 0; i < c.Len(); i++ {
+	for i := 0 ;i < c.Len(); i++ {
 		data = append(data, bool(c.values[i]))
 	}
 	fd.Field = &schema.FieldData_Scalars{
@@ -64,7 +73,7 @@ func (c *ColumnBool) ValueByIdx(idx int) (bool, error) {
 }
 
 // AppendValue append value into column
-func (c *ColumnBool) AppendValue(i interface{}) error {
+func(c *ColumnBool) AppendValue(i interface{}) error {
 	v, ok := i.(bool)
 	if !ok {
 		return fmt.Errorf("invalid type, expected bool, got %T", i)
@@ -81,8 +90,8 @@ func (c *ColumnBool) Data() []bool {
 
 // NewColumnBool auto generated constructor
 func NewColumnBool(name string, values []bool) *ColumnBool {
-	return &ColumnBool{
-		name:   name,
+	return &ColumnBool {
+		name: name,
 		values: values,
 	}
 }
@@ -108,14 +117,23 @@ func (c *ColumnInt8) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnInt8) Get(idx int) (interface{}, error) {
+	var r int8 // use default value
+	if idx < 0 || idx >= c.Len() {
+		return r, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnInt8) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Int8,
+		Type: schema.DataType_Int8,
 		FieldName: c.name,
 	}
 	data := make([]int32, 0, c.Len())
-	for i := 0; i < c.Len(); i++ {
+	for i := 0 ;i < c.Len(); i++ {
 		data = append(data, int32(c.values[i]))
 	}
 	fd.Field = &schema.FieldData_Scalars{
@@ -141,7 +159,7 @@ func (c *ColumnInt8) ValueByIdx(idx int) (int8, error) {
 }
 
 // AppendValue append value into column
-func (c *ColumnInt8) AppendValue(i interface{}) error {
+func(c *ColumnInt8) AppendValue(i interface{}) error {
 	v, ok := i.(int8)
 	if !ok {
 		return fmt.Errorf("invalid type, expected int8, got %T", i)
@@ -158,8 +176,8 @@ func (c *ColumnInt8) Data() []int8 {
 
 // NewColumnInt8 auto generated constructor
 func NewColumnInt8(name string, values []int8) *ColumnInt8 {
-	return &ColumnInt8{
-		name:   name,
+	return &ColumnInt8 {
+		name: name,
 		values: values,
 	}
 }
@@ -185,14 +203,23 @@ func (c *ColumnInt16) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnInt16) Get(idx int) (interface{}, error) {
+	var r int16 // use default value
+	if idx < 0 || idx >= c.Len() {
+		return r, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnInt16) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Int16,
+		Type: schema.DataType_Int16,
 		FieldName: c.name,
 	}
 	data := make([]int32, 0, c.Len())
-	for i := 0; i < c.Len(); i++ {
+	for i := 0 ;i < c.Len(); i++ {
 		data = append(data, int32(c.values[i]))
 	}
 	fd.Field = &schema.FieldData_Scalars{
@@ -218,7 +245,7 @@ func (c *ColumnInt16) ValueByIdx(idx int) (int16, error) {
 }
 
 // AppendValue append value into column
-func (c *ColumnInt16) AppendValue(i interface{}) error {
+func(c *ColumnInt16) AppendValue(i interface{}) error {
 	v, ok := i.(int16)
 	if !ok {
 		return fmt.Errorf("invalid type, expected int16, got %T", i)
@@ -235,8 +262,8 @@ func (c *ColumnInt16) Data() []int16 {
 
 // NewColumnInt16 auto generated constructor
 func NewColumnInt16(name string, values []int16) *ColumnInt16 {
-	return &ColumnInt16{
-		name:   name,
+	return &ColumnInt16 {
+		name: name,
 		values: values,
 	}
 }
@@ -262,14 +289,23 @@ func (c *ColumnInt32) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnInt32) Get(idx int) (interface{}, error) {
+	var r int32 // use default value
+	if idx < 0 || idx >= c.Len() {
+		return r, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnInt32) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Int32,
+		Type: schema.DataType_Int32,
 		FieldName: c.name,
 	}
 	data := make([]int32, 0, c.Len())
-	for i := 0; i < c.Len(); i++ {
+	for i := 0 ;i < c.Len(); i++ {
 		data = append(data, int32(c.values[i]))
 	}
 	fd.Field = &schema.FieldData_Scalars{
@@ -295,7 +331,7 @@ func (c *ColumnInt32) ValueByIdx(idx int) (int32, error) {
 }
 
 // AppendValue append value into column
-func (c *ColumnInt32) AppendValue(i interface{}) error {
+func(c *ColumnInt32) AppendValue(i interface{}) error {
 	v, ok := i.(int32)
 	if !ok {
 		return fmt.Errorf("invalid type, expected int32, got %T", i)
@@ -312,8 +348,8 @@ func (c *ColumnInt32) Data() []int32 {
 
 // NewColumnInt32 auto generated constructor
 func NewColumnInt32(name string, values []int32) *ColumnInt32 {
-	return &ColumnInt32{
-		name:   name,
+	return &ColumnInt32 {
+		name: name,
 		values: values,
 	}
 }
@@ -339,14 +375,23 @@ func (c *ColumnInt64) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnInt64) Get(idx int) (interface{}, error) {
+	var r int64 // use default value
+	if idx < 0 || idx >= c.Len() {
+		return r, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnInt64) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Int64,
+		Type: schema.DataType_Int64,
 		FieldName: c.name,
 	}
 	data := make([]int64, 0, c.Len())
-	for i := 0; i < c.Len(); i++ {
+	for i := 0 ;i < c.Len(); i++ {
 		data = append(data, int64(c.values[i]))
 	}
 	fd.Field = &schema.FieldData_Scalars{
@@ -372,7 +417,7 @@ func (c *ColumnInt64) ValueByIdx(idx int) (int64, error) {
 }
 
 // AppendValue append value into column
-func (c *ColumnInt64) AppendValue(i interface{}) error {
+func(c *ColumnInt64) AppendValue(i interface{}) error {
 	v, ok := i.(int64)
 	if !ok {
 		return fmt.Errorf("invalid type, expected int64, got %T", i)
@@ -389,8 +434,8 @@ func (c *ColumnInt64) Data() []int64 {
 
 // NewColumnInt64 auto generated constructor
 func NewColumnInt64(name string, values []int64) *ColumnInt64 {
-	return &ColumnInt64{
-		name:   name,
+	return &ColumnInt64 {
+		name: name,
 		values: values,
 	}
 }
@@ -416,14 +461,23 @@ func (c *ColumnFloat) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnFloat) Get(idx int) (interface{}, error) {
+	var r float32 // use default value
+	if idx < 0 || idx >= c.Len() {
+		return r, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnFloat) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Float,
+		Type: schema.DataType_Float,
 		FieldName: c.name,
 	}
 	data := make([]float32, 0, c.Len())
-	for i := 0; i < c.Len(); i++ {
+	for i := 0 ;i < c.Len(); i++ {
 		data = append(data, float32(c.values[i]))
 	}
 	fd.Field = &schema.FieldData_Scalars{
@@ -449,7 +503,7 @@ func (c *ColumnFloat) ValueByIdx(idx int) (float32, error) {
 }
 
 // AppendValue append value into column
-func (c *ColumnFloat) AppendValue(i interface{}) error {
+func(c *ColumnFloat) AppendValue(i interface{}) error {
 	v, ok := i.(float32)
 	if !ok {
 		return fmt.Errorf("invalid type, expected float32, got %T", i)
@@ -466,8 +520,8 @@ func (c *ColumnFloat) Data() []float32 {
 
 // NewColumnFloat auto generated constructor
 func NewColumnFloat(name string, values []float32) *ColumnFloat {
-	return &ColumnFloat{
-		name:   name,
+	return &ColumnFloat {
+		name: name,
 		values: values,
 	}
 }
@@ -493,14 +547,23 @@ func (c *ColumnDouble) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnDouble) Get(idx int) (interface{}, error) {
+	var r float64 // use default value
+	if idx < 0 || idx >= c.Len() {
+		return r, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnDouble) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Double,
+		Type: schema.DataType_Double,
 		FieldName: c.name,
 	}
 	data := make([]float64, 0, c.Len())
-	for i := 0; i < c.Len(); i++ {
+	for i := 0 ;i < c.Len(); i++ {
 		data = append(data, float64(c.values[i]))
 	}
 	fd.Field = &schema.FieldData_Scalars{
@@ -526,7 +589,7 @@ func (c *ColumnDouble) ValueByIdx(idx int) (float64, error) {
 }
 
 // AppendValue append value into column
-func (c *ColumnDouble) AppendValue(i interface{}) error {
+func(c *ColumnDouble) AppendValue(i interface{}) error {
 	v, ok := i.(float64)
 	if !ok {
 		return fmt.Errorf("invalid type, expected float64, got %T", i)
@@ -543,8 +606,8 @@ func (c *ColumnDouble) Data() []float64 {
 
 // NewColumnDouble auto generated constructor
 func NewColumnDouble(name string, values []float64) *ColumnDouble {
-	return &ColumnDouble{
-		name:   name,
+	return &ColumnDouble {
+		name: name,
 		values: values,
 	}
 }
@@ -570,14 +633,23 @@ func (c *ColumnString) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnString) Get(idx int) (interface{}, error) {
+	var r string // use default value
+	if idx < 0 || idx >= c.Len() {
+		return r, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnString) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_String,
+		Type: schema.DataType_String,
 		FieldName: c.name,
 	}
 	data := make([]string, 0, c.Len())
-	for i := 0; i < c.Len(); i++ {
+	for i := 0 ;i < c.Len(); i++ {
 		data = append(data, string(c.values[i]))
 	}
 	fd.Field = &schema.FieldData_Scalars{
@@ -603,7 +675,7 @@ func (c *ColumnString) ValueByIdx(idx int) (string, error) {
 }
 
 // AppendValue append value into column
-func (c *ColumnString) AppendValue(i interface{}) error {
+func(c *ColumnString) AppendValue(i interface{}) error {
 	v, ok := i.(string)
 	if !ok {
 		return fmt.Errorf("invalid type, expected string, got %T", i)
@@ -620,8 +692,9 @@ func (c *ColumnString) Data() []string {
 
 // NewColumnString auto generated constructor
 func NewColumnString(name string, values []string) *ColumnString {
-	return &ColumnString{
-		name:   name,
+	return &ColumnString {
+		name: name,
 		values: values,
 	}
 }
+
diff --git a/entity/columns_varchar.go b/entity/columns_varchar.go
index 2bffdd9..c000429 100644
--- a/entity/columns_varchar.go
+++ b/entity/columns_varchar.go
@@ -28,6 +28,14 @@ func (c *ColumnVarChar) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnVarChar) Get(idx int) (interface{}, error) {
+	if idx < 0 || idx > c.Len() {
+		return "", errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnVarChar) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
diff --git a/entity/columns_vector_gen.go b/entity/columns_vector_gen.go
index 7f7211a..7d83d17 100755
--- a/entity/columns_vector_gen.go
+++ b/entity/columns_vector_gen.go
@@ -1,14 +1,16 @@
 // Code generated by go generate; DO NOT EDIT
-// This file is generated by go genrated at 2022-01-26 15:50:36.315715425 +0800 CST m=+0.003539472
+// This file is generated by go generate 
 
-package entity
+package entity 
 
 import (
 	"fmt"
 
+	"github.com/cockroachdb/errors"
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 )
 
+
 // ColumnBinaryVector generated columns type for BinaryVector
 type ColumnBinaryVector struct {
 	name   string
@@ -27,7 +29,7 @@ func (c *ColumnBinaryVector) Type() FieldType {
 }
 
 // Len returns column data length
-func (c *ColumnBinaryVector) Len() int {
+func (c * ColumnBinaryVector) Len() int {
 	return len(c.values)
 }
 
@@ -36,8 +38,16 @@ func (c *ColumnBinaryVector) Dim() int {
 	return c.dim
 }
 
+// Get returns values at index as interface{}.
+func (c *ColumnBinaryVector) Get(idx int) (interface{}, error) {
+	if idx < 0 || idx >= c.Len() {
+		return nil, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // AppendValue append value into column
-func (c *ColumnBinaryVector) AppendValue(i interface{}) error {
+func(c *ColumnBinaryVector) AppendValue(i interface{}) error {
 	v, ok := i.([]byte)
 	if !ok {
 		return fmt.Errorf("invalid type, expected []byte, got %T", i)
@@ -55,11 +65,11 @@ func (c *ColumnBinaryVector) Data() [][]byte {
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnBinaryVector) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_BinaryVector,
+		Type: schema.DataType_BinaryVector,
 		FieldName: c.name,
 	}
 
-	data := make([]byte, 0, len(c.values)*c.dim)
+	data := make([]byte, 0, len(c.values)* c.dim)
 
 	for _, vector := range c.values {
 		data = append(data, vector...)
@@ -68,10 +78,11 @@ func (c *ColumnBinaryVector) FieldData() *schema.FieldData {
 	fd.Field = &schema.FieldData_Vectors{
 		Vectors: &schema.VectorField{
 			Dim: int64(c.dim),
-
+			
 			Data: &schema.VectorField_BinaryVector{
 				BinaryVector: data,
 			},
+			
 		},
 	}
 	return fd
@@ -79,7 +90,7 @@ func (c *ColumnBinaryVector) FieldData() *schema.FieldData {
 
 // NewColumnBinaryVector auto generated constructor
 func NewColumnBinaryVector(name string, dim int, values [][]byte) *ColumnBinaryVector {
-	return &ColumnBinaryVector{
+	return &ColumnBinaryVector {
 		name:   name,
 		dim:    dim,
 		values: values,
@@ -104,7 +115,7 @@ func (c *ColumnFloatVector) Type() FieldType {
 }
 
 // Len returns column data length
-func (c *ColumnFloatVector) Len() int {
+func (c * ColumnFloatVector) Len() int {
 	return len(c.values)
 }
 
@@ -113,8 +124,16 @@ func (c *ColumnFloatVector) Dim() int {
 	return c.dim
 }
 
+// Get returns values at index as interface{}.
+func (c *ColumnFloatVector) Get(idx int) (interface{}, error) {
+	if idx < 0 || idx >= c.Len() {
+		return nil, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // AppendValue append value into column
-func (c *ColumnFloatVector) AppendValue(i interface{}) error {
+func(c *ColumnFloatVector) AppendValue(i interface{}) error {
 	v, ok := i.([]float32)
 	if !ok {
 		return fmt.Errorf("invalid type, expected []float32, got %T", i)
@@ -132,11 +151,11 @@ func (c *ColumnFloatVector) Data() [][]float32 {
 // FieldData return column data mapped to schema.FieldData
 func (c *ColumnFloatVector) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
-		Type:      schema.DataType_FloatVector,
+		Type: schema.DataType_FloatVector,
 		FieldName: c.name,
 	}
 
-	data := make([]float32, 0, len(c.values)*c.dim)
+	data := make([]float32, 0, len(c.values)* c.dim)
 
 	for _, vector := range c.values {
 		data = append(data, vector...)
@@ -145,12 +164,13 @@ func (c *ColumnFloatVector) FieldData() *schema.FieldData {
 	fd.Field = &schema.FieldData_Vectors{
 		Vectors: &schema.VectorField{
 			Dim: int64(c.dim),
-
+			
 			Data: &schema.VectorField_FloatVector{
 				FloatVector: &schema.FloatArray{
 					Data: data,
 				},
 			},
+			
 		},
 	}
 	return fd
@@ -158,9 +178,10 @@ func (c *ColumnFloatVector) FieldData() *schema.FieldData {
 
 // NewColumnFloatVector auto generated constructor
 func NewColumnFloatVector(name string, dim int, values [][]float32) *ColumnFloatVector {
-	return &ColumnFloatVector{
+	return &ColumnFloatVector {
 		name:   name,
 		dim:    dim,
 		values: values,
 	}
 }
+
diff --git a/entity/gen/gen.go b/entity/gen/gen.go
index 5cdce42..70f4803 100644
--- a/entity/gen/gen.go
+++ b/entity/gen/gen.go
@@ -1,3 +1,4 @@
+//go:build ignore
 // +build ignore
 
 // Copyright (C) 2019-2021 Zilliz. All rights reserved.
@@ -24,15 +25,15 @@ import (
 )
 
 var scalarColumnTemplate = template.Must(template.New("").Parse(`// Code generated by go generate; DO NOT EDIT
-// This file is generated by go generate at {{ .Timestamp }}
+// This file is generated by go generate
 
 package entity 
 
 import (
-	"errors"
 	"fmt"
 
-	"github.com/milvus-io/milvus-sdk-go/v2/internal/proto/schema"
+	"github.com/cockroachdb/errors"
+	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 )
 {{ range .Types }}{{with .}}
 // Column{{.TypeName}} generated columns type for {{.TypeName}}
@@ -56,6 +57,15 @@ func (c *Column{{.TypeName}}) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *Column{{.TypeName}}) Get(idx int) (interface{}, error) {
+	var r {{.TypeDef}} // use default value
+	if idx < 0 || idx >= c.Len() {
+		return r, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData
 func (c *Column{{.TypeName}}) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
@@ -114,14 +124,15 @@ func NewColumn{{.TypeName}}(name string, values []{{.TypeDef}}) *Column{{.TypeNa
 {{end}}{{end}}
 `))
 var vectorColumnTemplate = template.Must(template.New("").Parse(`// Code generated by go generate; DO NOT EDIT
-// This file is generated by go genrated at {{ .Timestamp }}
+// This file is generated by go generate 
 
 package entity 
 
 import (
 	"fmt"
 
-	"github.com/milvus-io/milvus-sdk-go/v2/internal/proto/schema"
+	"github.com/cockroachdb/errors"
+	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 )
 
 {{ range .Types }}{{with.}}
@@ -152,6 +163,14 @@ func (c *Column{{.TypeName}}) Dim() int {
 	return c.dim
 }
 
+// Get returns values at index as interface{}.
+func (c *Column{{.TypeName}}) Get(idx int) (interface{}, error) {
+	if idx < 0 || idx >= c.Len() {
+		return nil, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // AppendValue append value into column
 func(c *Column{{.TypeName}}) AppendValue(i interface{}) error {
 	v, ok := i.({{.TypeDef}})
@@ -212,7 +231,7 @@ func NewColumn{{.TypeName}}(name string, dim int, values []{{.TypeDef}}) *Column
 `))
 
 var scalarColumnTestTemplate = template.Must(template.New("").Parse(`// Code generated by go generate; DO NOT EDIT
-// This file is generated by go genrated at {{ .Timestamp }}
+// This file is generated by go generate 
 
 package entity 
 
@@ -223,7 +242,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/milvus-io/milvus-sdk-go/v2/internal/proto/schema"
+	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 )
 {{ range .Types }}{{with.}}
 func TestColumn{{.TypeName}}(t *testing.T) {
@@ -334,7 +353,7 @@ func TestFieldData{{.TypeName}}Column(t *testing.T) {
 {{end}}{{end}}
 `))
 var vectorColumnTestTemplate = template.Must(template.New("").Parse(`// Code generated by go generate; DO NOT EDIT
-// This file is generated by go genrated at {{ .Timestamp }}
+// This file is generated by go generate 
 
 package entity 
 
@@ -344,7 +363,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/milvus-io/milvus-sdk-go/v2/internal/proto/schema"
+	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 	"github.com/stretchr/testify/assert"
 )
 {{range .Types}}{{with .}}
diff --git a/entity/rows.go b/entity/rows.go
index 8e06726..4c7d1ca 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,10 @@ 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)
+			nameColumns[field.Name] = col
 		case FieldTypeFloatVector:
 			data := make([][]float32, 0, rowsLen)
 			dimStr, has := field.TypeParams[TypeParamDim]
@@ -297,6 +313,11 @@ func RowsToColumns(rows []Row, schemas ...*Schema) ([]Column, error) {
 			nameColumns[field.Name] = col
 		}
 	}
+
+	if isDynamic {
+		dynamicCol = NewColumnJSONBytes("", make([][]byte, 0, rowsLen)).WithIsDynamic(true)
+	}
+
 	for _, row := range rows {
 		// collection schema name need not to be same, since receiver could has other names
 		v := reflect.ValueOf(row)
@@ -304,43 +325,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 isDynamic {
+		columns = append(columns, dynamicCol)
 	}
-	// try use name directly
-	return v.FieldByName(name)
+	return columns, nil
 }
diff --git a/entity/schema.go b/entity/schema.go
index a412aed..e46310d 100644
--- a/entity/schema.go
+++ b/entity/schema.go
@@ -12,6 +12,8 @@
 package entity
 
 import (
+	"strconv"
+
 	common "github.com/milvus-io/milvus-proto/go-api/commonpb"
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 )
@@ -45,18 +47,53 @@ func (cl ConsistencyLevel) CommonConsistencyLevel() common.ConsistencyLevel {
 
 // Schema represents schema info of collection in milvus
 type Schema struct {
-	CollectionName string
-	Description    string
-	AutoID         bool
-	Fields         []*Field
+	CollectionName     string
+	Description        string
+	AutoID             bool
+	Fields             []*Field
+	EnableDynamicField bool
+}
+
+// NewSchema creates an empty schema object.
+func NewSchema() *Schema {
+	return &Schema{}
+}
+
+// WithName sets the name value of schema, returns schema itself.
+func (s *Schema) WithName(name string) *Schema {
+	s.CollectionName = name
+	return s
+}
+
+// WithDescription sets the description value of schema, returns schema itself.
+func (s *Schema) WithDescription(desc string) *Schema {
+	s.Description = desc
+	return s
+}
+
+func (s *Schema) WithAutoID(autoID bool) *Schema {
+	s.AutoID = autoID
+	return s
+}
+
+func (s *Schema) WithDynamicFieldEnabled(dynamicEnabled bool) *Schema {
+	s.EnableDynamicField = dynamicEnabled
+	return s
+}
+
+// WithField adds a field into schema and returns schema itself.
+func (s *Schema) WithField(f *Field) *Schema {
+	s.Fields = append(s.Fields, f)
+	return s
 }
 
 // ProtoMessage returns corresponding server.CollectionSchema
 func (s *Schema) ProtoMessage() *schema.CollectionSchema {
 	r := &schema.CollectionSchema{
-		Name:        s.CollectionName,
-		Description: s.Description,
-		AutoID:      s.AutoID,
+		Name:               s.CollectionName,
+		Description:        s.Description,
+		AutoID:             s.AutoID,
+		EnableDynamicField: s.EnableDynamicField,
 	}
 	r.Fields = make([]*schema.FieldSchema, 0, len(s.Fields))
 	for _, field := range s.Fields {
@@ -72,35 +109,117 @@ func (s *Schema) ReadProto(p *schema.CollectionSchema) *Schema {
 	s.CollectionName = p.GetName()
 	s.Fields = make([]*Field, 0, len(p.GetFields()))
 	for _, fp := range p.GetFields() {
-		s.Fields = append(s.Fields, (&Field{}).ReadProto(fp))
+		s.Fields = append(s.Fields, NewField().ReadProto(fp))
 	}
+	s.EnableDynamicField = p.GetEnableDynamicField()
 	return s
 }
 
+// PKFieldName returns pk field name for this schema.
+func (s *Schema) PKFieldName() string {
+	for _, field := range s.Fields {
+		if field.PrimaryKey {
+			return field.Name
+		}
+	}
+	return ""
+}
+
 // Field represent field schema in milvus
 type Field struct {
-	ID          int64  // field id, generated when collection is created, input value is ignored
-	Name        string // field name
-	PrimaryKey  bool   // is primary key
-	AutoID      bool   // is auto id
-	Description string
-	DataType    FieldType
-	TypeParams  map[string]string
-	IndexParams map[string]string
+	ID             int64  // field id, generated when collection is created, input value is ignored
+	Name           string // field name
+	PrimaryKey     bool   // is primary key
+	AutoID         bool   // is auto id
+	Description    string
+	DataType       FieldType
+	TypeParams     map[string]string
+	IndexParams    map[string]string
+	IsDynamic      bool
+	IsPartitionKey bool
 }
 
 // ProtoMessage generates corresponding FieldSchema
 func (f *Field) ProtoMessage() *schema.FieldSchema {
 	return &schema.FieldSchema{
-		FieldID:      f.ID,
-		Name:         f.Name,
-		Description:  f.Description,
-		IsPrimaryKey: f.PrimaryKey,
-		AutoID:       f.AutoID,
-		DataType:     schema.DataType(f.DataType),
-		TypeParams:   MapKvPairs(f.TypeParams),
-		IndexParams:  MapKvPairs(f.IndexParams),
+		FieldID:        f.ID,
+		Name:           f.Name,
+		Description:    f.Description,
+		IsPrimaryKey:   f.PrimaryKey,
+		AutoID:         f.AutoID,
+		DataType:       schema.DataType(f.DataType),
+		TypeParams:     MapKvPairs(f.TypeParams),
+		IndexParams:    MapKvPairs(f.IndexParams),
+		IsDynamic:      f.IsDynamic,
+		IsPartitionKey: f.IsPartitionKey,
+	}
+}
+
+// NewField creates a new Field with map initialized.
+func NewField() *Field {
+	return &Field{
+		TypeParams:  make(map[string]string),
+		IndexParams: make(map[string]string),
+	}
+}
+
+func (f *Field) WithName(name string) *Field {
+	f.Name = name
+	return f
+}
+
+func (f *Field) WithDescription(desc string) *Field {
+	f.Description = desc
+	return f
+}
+
+func (f *Field) WithDataType(dataType FieldType) *Field {
+	f.DataType = dataType
+	return f
+}
+
+func (f *Field) WithIsPrimaryKey(isPrimaryKey bool) *Field {
+	f.PrimaryKey = isPrimaryKey
+	return f
+}
+
+func (f *Field) WithIsAutoID(isAutoID bool) *Field {
+	f.AutoID = isAutoID
+	return f
+}
+
+func (f *Field) WithIsDynamic(isDynamic bool) *Field {
+	f.IsDynamic = isDynamic
+	return f
+}
+
+func (f *Field) WithIsPartitionKey(isPartitionKey bool) *Field {
+	f.IsPartitionKey = isPartitionKey
+	return f
+}
+
+func (f *Field) WithTypeParams(key string, value string) *Field {
+	if f.TypeParams == nil {
+		f.TypeParams = make(map[string]string)
 	}
+	f.TypeParams[key] = value
+	return f
+}
+
+func (f *Field) WithDim(dim int64) *Field {
+	if f.TypeParams == nil {
+		f.TypeParams = make(map[string]string)
+	}
+	f.TypeParams[TypeParamDim] = strconv.FormatInt(dim, 10)
+	return f
+}
+
+func (f *Field) WithMaxLength(maxLen int64) *Field {
+	if f.TypeParams == nil {
+		f.TypeParams = make(map[string]string)
+	}
+	f.TypeParams[TypeParamMaxLength] = strconv.FormatInt(maxLen, 10)
+	return f
 }
 
 // ReadProto parses FieldSchema
@@ -113,6 +232,8 @@ func (f *Field) ReadProto(p *schema.FieldSchema) *Field {
 	f.DataType = FieldType(p.GetDataType())
 	f.TypeParams = KvPairsMap(p.GetTypeParams())
 	f.IndexParams = KvPairsMap(p.GetIndexParams())
+	f.IsDynamic = p.GetIsDynamic()
+	f.IsPartitionKey = p.GetIsPartitionKey()
 
 	return f
 }
@@ -235,9 +356,7 @@ func (t FieldType) PbFieldType() (string, string) {
 		return "[]float32", ""
 	default:
 		return "undefined", ""
-
 	}
-
 }
 
 // Match schema definition
diff --git a/go.mod b/go.mod
index a14dc88..abfa646 100644
--- a/go.mod
+++ b/go.mod
@@ -13,6 +13,12 @@ require (
 	google.golang.org/grpc/examples v0.0.0-20220617181431-3e7b97febc7f
 )
 
+require (
+	github.com/tidwall/gjson v1.14.4 // indirect
+	github.com/tidwall/match v1.1.1 // indirect
+	github.com/tidwall/pretty v1.2.0 // indirect
+)
+
 require (
 	github.com/cockroachdb/logtags v0.0.0-20211118104740-dabe8e521a4f // indirect
 	github.com/cockroachdb/redact v1.1.3 // indirect
diff --git a/go.sum b/go.sum
index bce7e80..554f18b 100644
--- a/go.sum
+++ b/go.sum
@@ -218,6 +218,12 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/tidwall/gjson v1.14.4 h1:uo0p8EbA09J7RQaflQ1aBRffTR7xedD2bcIVSYxLnkM=
+github.com/tidwall/gjson v1.14.4/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
+github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
+github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
 github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=


Test patch:
diff --git a/client/client_mock_test.go b/client/client_mock_test.go
index 8208f3e..0949459 100644
--- a/client/client_mock_test.go
+++ b/client/client_mock_test.go
@@ -73,6 +73,7 @@ func (s *MockSuiteBase) TearDownTest() {
 }
 
 func (s *MockSuiteBase) resetMock() {
+	MetaCache.reset()
 	if s.mock != nil {
 		s.mock.Calls = nil
 		s.mock.ExpectedCalls = nil
@@ -172,7 +173,7 @@ func (s *MockSuiteBase) getVarcharFieldData(name string, data []string) *schema.
 	}
 }
 
-func (s *MockSuiteBase) getJSONBytesFieldData(name string, data [][]byte) *schema.FieldData {
+func (s *MockSuiteBase) getJSONBytesFieldData(name string, data [][]byte, isDynamic bool) *schema.FieldData {
 	return &schema.FieldData{
 		Type:      schema.DataType_JSON,
 		FieldName: name,
@@ -185,6 +186,7 @@ func (s *MockSuiteBase) getJSONBytesFieldData(name string, data [][]byte) *schem
 				},
 			},
 		},
+		IsDynamic: isDynamic,
 	}
 }
 
@@ -909,6 +911,10 @@ func (m *MockServer) Connect(_ context.Context, _ *server.ConnectRequest) (*serv
 	panic("not implemented") // TODO: Implement
 }
 
+func getSuccessStatus() *common.Status {
+	return &common.Status{ErrorCode: common.ErrorCode_Success}
+}
+
 func SuccessStatus() (*common.Status, error) {
 	return &common.Status{ErrorCode: common.ErrorCode_Success}, nil
 }
diff --git a/client/data_test.go b/client/data_test.go
index 60a28d3..4c54328 100644
--- a/client/data_test.go
+++ b/client/data_test.go
@@ -1,113 +1,36 @@
+// Copyright (C) 2019-2021 Zilliz. All rights reserved.
+//
+// Licensed 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 client
 
 import (
 	"context"
-	"errors"
-	"fmt"
 	"math/rand"
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/errors"
+
 	"github.com/golang/protobuf/proto"
-	"github.com/milvus-io/milvus-proto/go-api/commonpb"
 	common "github.com/milvus-io/milvus-proto/go-api/commonpb"
 	server "github.com/milvus-io/milvus-proto/go-api/milvuspb"
 	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/require"
+	"github.com/stretchr/testify/suite"
 )
 
-func TestGrpcClientInsert(t *testing.T) {
-	ctx := context.Background()
-
-	c := testClient(ctx, t)
-
-	t.Run("test create failure due to meta", func(t *testing.T) {
-		mockServer.DelInjection(MHasCollection) // collection does not exist
-		ids, err := c.Insert(ctx, testCollectionName, "")
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-
-		// partition not exists
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		ids, err = c.Insert(ctx, testCollectionName, "_part_not_exists")
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-
-		// field not in collection
-		mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, 0, testCollectionName, defaultSchema()))
-		vectors := generateFloatVector(10, testVectorDim)
-		ids, err = c.Insert(ctx, testCollectionName, "",
-			entity.NewColumnInt64("extra_field", []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}), entity.NewColumnFloatVector(testVectorField, testVectorDim, vectors))
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-
-		// field type not match
-		mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, 0, testCollectionName, defaultSchema()))
-		ids, err = c.Insert(ctx, testCollectionName, "",
-			entity.NewColumnInt32("int64", []int32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}), entity.NewColumnFloatVector(testVectorField, testVectorDim, vectors))
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-
-		// missing field
-		ids, err = c.Insert(ctx, testCollectionName, "")
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-
-		// column len not match
-		ids, err = c.Insert(ctx, testCollectionName, "", entity.NewColumnInt64("int64", []int64{1, 2, 3, 4, 5, 6, 7, 8, 9}),
-			entity.NewColumnFloatVector(testVectorField, testVectorDim, vectors))
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-
-		// column len not match
-		ids, err = c.Insert(ctx, testCollectionName, "", entity.NewColumnInt64("int64", []int64{1, 2, 3}),
-			entity.NewColumnFloatVector(testVectorField, testVectorDim, vectors))
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-
-		// dim not match
-		ids, err = c.Insert(ctx, testCollectionName, "", entity.NewColumnInt64("int64", []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}),
-			entity.NewColumnFloatVector(testVectorField, testVectorDim*2, vectors))
-		assert.Nil(t, ids)
-		assert.NotNil(t, err)
-	})
-
-	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, err := SuccessStatus()
-		resp.Status = s
-		return resp, err
-	})
-	_, err := c.Insert(ctx, testCollectionName, "", // use default partition
-		entity.NewColumnFloatVector(testVectorField, testVectorDim, vector))
-	assert.Nil(t, err)
-	mockServer.DelInjection(MInsert)
-}
-
 func TestGrpcClientFlush(t *testing.T) {
-
 	ctx := context.Background()
 
 	c := testClient(ctx, t)
@@ -238,7 +161,7 @@ func TestGrpcDeleteByPks(t *testing.T) {
 	t.Run("delete services fail", func(t *testing.T) {
 		mockServer.SetInjection(MDelete, func(_ context.Context, raw proto.Message) (proto.Message, error) {
 			resp := &server.MutationResult{}
-			return resp, errors.New("mocked error")
+			return resp, errors.New("mockServer.d error")
 		})
 
 		err := c.DeleteByPks(ctx, testCollectionName, "", entity.NewColumnInt64(testPrimaryField, []int64{1}))
@@ -256,667 +179,435 @@ func TestGrpcDeleteByPks(t *testing.T) {
 	})
 }
 
-func TestGrpcSearch(t *testing.T) {
-
-	ctx := context.Background()
+type SearchSuite struct {
+	MockSuiteBase
+	sch        *entity.Schema
+	schDynamic *entity.Schema
+}
 
-	c := testClient(ctx, t)
-	defer c.Close()
-	vectors := generateFloatVector(4096, testVectorDim)
+func (s *SearchSuite) SetupSuite() {
+	s.MockSuiteBase.SetupSuite()
 
-	t.Run("search fail due to meta error", func(t *testing.T) {
-		sp, err := entity.NewIndexFlatSearchParam()
-		assert.Nil(t, err)
+	s.sch = entity.NewSchema().WithName(testCollectionName).
+		WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
+		WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithDim(testVectorDim))
+	s.schDynamic = entity.NewSchema().WithName(testCollectionName).WithDynamicFieldEnabled(true).
+		WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
+		WithField(entity.NewField().WithName("$meta").WithDataType(entity.FieldTypeJSON).WithIsDynamic(true)).
+		WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithDim(testVectorDim))
+}
 
-		// specify guarantee timestamp in strong consistency level
-		r, err := c.Search(ctx, testCollectionName, []string{}, "", []string{"int64"}, []entity.Vector{entity.FloatVector(vectors[0])}, "vector",
-			entity.HAMMING, 5, sp, WithSearchQueryConsistencyLevel(entity.ClStrong), WithGuaranteeTimestamp(1))
-		assert.Nil(t, r)
-		assert.NotNil(t, err)
+func (s *SearchSuite) TestSearchFail() {
+	c := s.client
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	partName := "part_1"
+	vectors := generateFloatVector(10, testVectorDim)
+	sp, err := entity.NewIndexFlatSearchParam()
+	s.Require().NoError(err)
+	s.resetMock()
+
+	s.Run("service_not_ready", func() {
+		_, err := (&GrpcClient{}).Search(ctx, testCollectionName, []string{}, partName, []string{"ID"}, []entity.Vector{entity.FloatVector(vectors[0])}, "vector",
+			entity.L2, 5, sp, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
+		s.ErrorIs(err, ErrClientNotReady)
 	})
 
-	t.Run("ok search", func(t *testing.T) {
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, 0, testCollectionName, defaultSchema()))
+	s.Run("fail_describecollection_error", func() {
+		defer s.resetMock()
 
-		expr := `int64 > 0`
+		s.setupDescribeCollectionError(common.ErrorCode_Success, errors.New("mock error"))
 
-		mockServer.SetInjection(MSearch, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			req, ok := raw.(*server.SearchRequest)
-			resp := &server.SearchResults{}
-			if !ok {
-				s, err := BadRequestStatus()
-				resp.Status = s
-				return resp, err
-			}
-			assert.Equal(t, testCollectionName, req.GetCollectionName())
-			assert.Equal(t, expr, req.GetDsl())
-			assert.Equal(t, common.DslType_BoolExprV1, req.GetDslType())
-			assert.ElementsMatch(t, []string{"int64"}, req.GetOutputFields())
+		_, err := c.Search(ctx, testCollectionName, []string{partName}, "", []string{"ID"}, []entity.Vector{entity.FloatVector(vectors[0])}, "vector",
+			entity.L2, 5, sp, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
+	})
 
-			resp.Results = &schema.SearchResultData{
-				NumQueries: 1,
-				TopK:       10,
-				FieldsData: []*schema.FieldData{
-					{
-						Type:      schema.DataType_Int64,
-						FieldName: "int64",
-						Field: &schema.FieldData_Scalars{
-							Scalars: &schema.ScalarField{
-								Data: &schema.ScalarField_LongData{
-									LongData: &schema.LongArray{
-										Data: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
-									},
-								},
-							},
-						},
-					},
-				},
-				Ids: &schema.IDs{
-					IdField: &schema.IDs_IntId{
-						IntId: &schema.LongArray{
-							Data: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
-						},
-					},
-				},
-				Scores: make([]float32, 10),
-				Topks:  []int64{10},
-			}
+	s.Run("fail_describecollection_errcode", func() {
+		defer s.resetMock()
 
-			s, err := SuccessStatus()
-			resp.Status = s
-			return resp, err
-		})
+		s.setupDescribeCollectionError(common.ErrorCode_UnexpectedError, nil)
 
-		sp, err := entity.NewIndexFlatSearchParam()
-		assert.Nil(t, err)
-		assert.NotNil(t, sp)
-		results, err := c.Search(ctx, testCollectionName, []string{}, expr, []string{"int64"}, []entity.Vector{entity.FloatVector(vectors[0])},
-			testVectorField, entity.L2, 10, sp)
+		_, err := c.Search(ctx, testCollectionName, []string{partName}, "", []string{"ID"}, []entity.Vector{entity.FloatVector(vectors[0])}, "vector",
+			entity.L2, 5, sp, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
+	})
 
-		assert.Nil(t, err)
-		assert.NotNil(t, results)
+	s.Run("fail_guaranteed_non_custom_cl", func() {
+		defer s.resetMock()
 
-		// search with ignore growing
-		results, err = c.Search(ctx, testCollectionName, []string{}, expr, []string{"int64"}, []entity.Vector{entity.FloatVector(vectors[0])},
-			testVectorField, entity.L2, 10, sp, WithIgnoreGrowing())
-		assert.Nil(t, err)
-		assert.NotNil(t, results)
+		s.setupDescribeCollection(testCollectionName, s.sch)
 
-		// search with session consistency level
-		results, err = c.Search(ctx, testCollectionName, []string{}, expr, []string{"int64"}, []entity.Vector{entity.FloatVector(vectors[0])},
-			testVectorField, entity.L2, 10, sp, WithSearchQueryConsistencyLevel(entity.ClSession), WithTravelTimestamp(0))
-		assert.Nil(t, err)
-		assert.NotNil(t, results)
+		_, err := c.Search(ctx, testCollectionName, []string{partName}, "", []string{"ID"}, []entity.Vector{entity.FloatVector(vectors[0])}, "vector",
+			entity.L2, 5, sp, WithSearchQueryConsistencyLevel(entity.ClStrong), WithGuaranteeTimestamp(1000000))
+		s.Error(err)
+	})
 
-		// search with customized consistency level
-		results, err = c.Search(ctx, testCollectionName, []string{}, expr, []string{"int64"}, []entity.Vector{entity.FloatVector(vectors[0])},
-			testVectorField, entity.L2, 10, sp, WithSearchQueryConsistencyLevel(entity.ClCustomized), WithTravelTimestamp(1))
-		assert.Nil(t, err)
-		assert.NotNil(t, results)
+	s.Run("fail_search_error", func() {
+		defer s.resetMock()
+
+		s.setupDescribeCollection(testCollectionName, s.sch)
+		s.mock.EXPECT().Search(mock.Anything, mock.AnythingOfType("*milvuspb.SearchRequest")).
+			Return(nil, errors.New("mock error"))
+
+		_, err := c.Search(ctx, testCollectionName, []string{partName}, "", []string{"ID"}, []entity.Vector{entity.FloatVector(vectors[0])}, "vector",
+			entity.L2, 5, sp, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
 	})
-}
 
-func TestGrpcQuery(t *testing.T) {
-	ctx := context.Background()
-	c := testClient(ctx, t)
-	partName := "testPart"
+	s.Run("fail_search_errcode", func() {
+		defer s.resetMock()
 
-	t.Run("normal query", func(t *testing.T) {
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		mockServer.SetInjection(MHasPartition, hasPartitionInjection(t, testCollectionName, false, partName))
-		defer mockServer.DelInjection(MHasPartition)
+		s.setupDescribeCollection(testCollectionName, s.sch)
+		s.mock.EXPECT().Search(mock.Anything, mock.AnythingOfType("*milvuspb.SearchRequest")).
+			Return(&server.SearchResults{Status: &common.Status{ErrorCode: common.ErrorCode_UnexpectedError}}, nil)
 
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			req, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
-			assert.Equal(t, testCollectionName, req.GetCollectionName())
-			if len(req.GetPartitionNames()) > 0 {
-				assert.Equal(t, partName, req.GetPartitionNames()[0])
-			}
+		_, err := c.Search(ctx, testCollectionName, []string{partName}, "", []string{"ID"}, []entity.Vector{entity.FloatVector(vectors[0])}, "vector",
+			entity.L2, 5, sp, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
+	})
+}
 
-			resp := &server.QueryResults{}
-			s, err := SuccessStatus()
-			resp.Status = s
-			resp.FieldsData = []*schema.FieldData{
-				{
-					Type:      schema.DataType_Int64,
-					FieldName: "int64",
-					Field: &schema.FieldData_Scalars{
-						Scalars: &schema.ScalarField{
-							Data: &schema.ScalarField_LongData{
-								LongData: &schema.LongArray{
-									Data: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
-								},
-							},
-						},
+func (s *SearchSuite) TestSearchSuccess() {
+	c := s.client
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	partName := "part_1"
+	vectors := generateFloatVector(10, testVectorDim)
+	sp, err := entity.NewIndexFlatSearchParam()
+	s.Require().NoError(err)
+	s.resetMock()
+
+	expr := "ID > 0"
+
+	s.Run("non_dynamic_schema", func() {
+		defer s.resetMock()
+		s.setupDescribeCollection(testCollectionName, s.sch)
+		s.mock.EXPECT().Search(mock.Anything, mock.AnythingOfType("*milvuspb.SearchRequest")).
+			Run(func(_ context.Context, req *server.SearchRequest) {
+				s.Equal(testCollectionName, req.GetCollectionName())
+				s.Equal(expr, req.GetDsl())
+				s.Equal(common.DslType_BoolExprV1, req.GetDslType())
+				s.ElementsMatch([]string{"ID"}, req.GetOutputFields())
+				s.ElementsMatch([]string{partName}, req.GetPartitionNames())
+			}).
+			Return(&server.SearchResults{
+				Status: getSuccessStatus(),
+				Results: &schema.SearchResultData{
+					NumQueries: 1,
+					TopK:       10,
+					FieldsData: []*schema.FieldData{
+						s.getInt64FieldData("ID", []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}),
 					},
-				},
-				{
-					Type:      schema.DataType_FloatVector,
-					FieldName: testVectorField,
-					Field: &schema.FieldData_Vectors{
-						Vectors: &schema.VectorField{
-							Dim: 1,
-							Data: &schema.VectorField_FloatVector{
-								FloatVector: &schema.FloatArray{
-									Data: []float32{0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0},
-								},
+					Ids: &schema.IDs{
+						IdField: &schema.IDs_IntId{
+							IntId: &schema.LongArray{
+								Data: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
 							},
 						},
 					},
+					Scores: make([]float32, 10),
+					Topks:  []int64{10},
 				},
-			}
-
-			return resp, err
-		})
-		defer mockServer.DelInjection(MQuery)
-
-		columns, err := c.Query(ctx, testCollectionName, []string{partName}, "int64 in {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}", []string{testPrimaryField, testVectorField}, WithOffset(1), WithLimit(10))
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(columns))
-		assert.Equal(t, entity.FieldTypeInt64, columns[0].Type())
-		assert.Equal(t, entity.FieldTypeFloatVector, columns[1].Type())
-		assert.Equal(t, 10, columns[0].Len())
-
-		colInt64, ok := columns[0].(*entity.ColumnInt64)
-		assert.True(t, ok)
-		assert.ElementsMatch(t, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, colInt64.Data())
-
-		// query with options
-		columns, err = c.Query(ctx, testCollectionName, []string{partName}, "int64 in {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}", []string{"int64", testVectorField}, WithSearchQueryConsistencyLevel(entity.ClSession), WithTravelTimestamp(0))
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(columns))
-		assert.Equal(t, entity.FieldTypeInt64, columns[0].Type())
-		assert.Equal(t, entity.FieldTypeFloatVector, columns[1].Type())
-		assert.Equal(t, 10, columns[0].Len())
-
-		colInt64, ok = columns[0].(*entity.ColumnInt64)
-		assert.True(t, ok)
-		assert.ElementsMatch(t, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, colInt64.Data())
-
-		// query with ignore growing options
-		columns, err = c.Query(ctx, testCollectionName, []string{partName}, "int64 in {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}", []string{"int64", testVectorField}, WithIgnoreGrowing())
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(columns))
-		assert.Equal(t, entity.FieldTypeInt64, columns[0].Type())
-		assert.Equal(t, entity.FieldTypeFloatVector, columns[1].Type())
-		assert.Equal(t, 10, columns[0].Len())
-
-		colInt64, ok = columns[0].(*entity.ColumnInt64)
-		assert.True(t, ok)
-		assert.ElementsMatch(t, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, colInt64.Data())
+			}, nil)
+
+		r, err := c.Search(ctx, testCollectionName, []string{partName}, expr, []string{"ID"}, []entity.Vector{entity.FloatVector(vectors[0])},
+			testVectorField, entity.L2, 10, sp, WithIgnoreGrowing(), WithSearchQueryConsistencyLevel(entity.ClCustomized), WithGuaranteeTimestamp(10000000000))
+		s.NoError(err)
+		s.Require().Equal(1, len(r))
+		result := r[0]
+		s.Require().NotNil(result.Fields.GetColumn("ID"))
 	})
 
-	t.Run("normal query varchar pks", func(t *testing.T) {
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		mockServer.SetInjection(MHasPartition, hasPartitionInjection(t, testCollectionName, false, partName))
-		defer mockServer.DelInjection(MHasPartition)
-		mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, testCollectionID, testCollectionName, varCharSchema()))
-		defer mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, testCollectionID, testCollectionName, defaultSchema()))
-
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			req, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
-			assert.Equal(t, testCollectionName, req.GetCollectionName())
-			if len(req.GetPartitionNames()) > 0 {
-				assert.Equal(t, partName, req.GetPartitionNames()[0])
-			}
-
-			resp := &server.QueryResults{}
-			s, err := SuccessStatus()
-			resp.Status = s
-			resp.FieldsData = []*schema.FieldData{
-				{
-					Type:      schema.DataType_VarChar,
-					FieldName: "varchar",
-					Field: &schema.FieldData_Scalars{
-						Scalars: &schema.ScalarField{
-							Data: &schema.ScalarField_StringData{
-								StringData: &schema.StringArray{
-									Data: []string{"1", "2", "3", "4", "5", "6", "7", "8", "9", "10"},
-								},
-							},
-						},
+	s.Run("dynamic_schema", func() {
+		defer s.resetMock()
+		s.setupDescribeCollection(testCollectionName, s.schDynamic)
+		s.mock.EXPECT().Search(mock.Anything, mock.AnythingOfType("*milvuspb.SearchRequest")).
+			Run(func(_ context.Context, req *server.SearchRequest) {
+				s.Equal(testCollectionName, req.GetCollectionName())
+				s.Equal(expr, req.GetDsl())
+				s.Equal(common.DslType_BoolExprV1, req.GetDslType())
+				s.ElementsMatch([]string{"A", "B"}, req.GetOutputFields())
+				s.ElementsMatch([]string{partName}, req.GetPartitionNames())
+			}).
+			Return(&server.SearchResults{
+				Status: getSuccessStatus(),
+				Results: &schema.SearchResultData{
+					NumQueries: 1,
+					TopK:       2,
+					FieldsData: []*schema.FieldData{
+						s.getJSONBytesFieldData("", [][]byte{
+							[]byte(`{"A": 123, "B": "456"}`),
+							[]byte(`{"B": "abc", "A": 456}`),
+						}, true),
 					},
-				},
-				{
-					Type:      schema.DataType_FloatVector,
-					FieldName: testVectorField,
-					Field: &schema.FieldData_Vectors{
-						Vectors: &schema.VectorField{
-							Dim: 1,
-							Data: &schema.VectorField_FloatVector{
-								FloatVector: &schema.FloatArray{
-									Data: []float32{0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0},
-								},
+					Ids: &schema.IDs{
+						IdField: &schema.IDs_IntId{
+							IntId: &schema.LongArray{
+								Data: []int64{1, 2},
 							},
 						},
 					},
+					Scores: make([]float32, 2),
+					Topks:  []int64{2},
 				},
-			}
-
-			return resp, err
-		})
-		defer mockServer.DelInjection(MQuery)
-
-		columns, err := c.Query(ctx, testCollectionName, []string{partName}, `varchar in {"1", "2", "3", "4", "5", "6", "7", "8", "9", "10"}`, []string{"varchar", testVectorField})
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(columns))
-		assert.Equal(t, entity.FieldTypeVarChar, columns[0].Type())
-		assert.Equal(t, entity.FieldTypeFloatVector, columns[1].Type())
-		assert.Equal(t, 10, columns[0].Len())
-
-		colPK, ok := columns[0].(*entity.ColumnVarChar)
-		assert.True(t, ok)
-		assert.ElementsMatch(t, []string{"1", "2", "3", "4", "5", "6", "7", "8", "9", "10"}, colPK.Data())
+			}, nil)
+
+		r, err := c.Search(ctx, testCollectionName, []string{partName}, expr, []string{"A", "B"}, []entity.Vector{entity.FloatVector(vectors[0])},
+			testVectorField, entity.L2, 2, sp, WithIgnoreGrowing(), WithSearchQueryConsistencyLevel(entity.ClBounded))
+		s.NoError(err)
+		s.Require().Equal(1, len(r))
+		result := r[0]
+		columnA := result.Fields.GetColumn("A")
+		s.Require().NotNil(columnA)
+		column, ok := columnA.(*entity.ColumnDynamic)
+		s.Require().True(ok)
+		v, err := column.GetInt64(0)
+		s.NoError(err)
+		s.Equal(int64(123), v)
+
+		columnB := result.Fields.GetColumn("B")
+		s.Require().NotNil(columnB)
+		column, ok = columnB.(*entity.ColumnDynamic)
+		s.Require().True(ok)
+		str, err := column.GetString(1)
+		s.NoError(err)
+		s.Equal("abc", str)
 	})
+}
 
-	t.Run("Bad request querys", func(t *testing.T) {
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		mockServer.SetInjection(MHasPartition, hasPartitionInjection(t, testCollectionName, false, partName))
-		defer mockServer.DelInjection(MHasPartition)
-		mockServer.SetInjection(MDescribeCollection, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			req, ok := raw.(*server.DescribeCollectionRequest)
-			resp := &server.DescribeCollectionResponse{}
-			if !ok {
-				s, err := BadRequestStatus()
-				resp.Status = s
-				return resp, err
-			}
+func TestSearch(t *testing.T) {
+	suite.Run(t, new(SearchSuite))
+}
 
-			if req.CollectionName != testCollectionName {
-				resp.Status = &commonpb.Status{
-					ErrorCode: commonpb.ErrorCode_CollectionNameNotFound,
-				}
-				return resp, nil
-			}
-			sch := defaultSchema()
-			resp.Schema = sch.ProtoMessage()
-			resp.CollectionID = testCollectionID
+type QuerySuite struct {
+	MockSuiteBase
+	sch        *entity.Schema
+	schDynamic *entity.Schema
+}
 
-			s, err := SuccessStatus()
-			resp.Status = s
+func (s *QuerySuite) SetupSuite() {
+	s.MockSuiteBase.SetupSuite()
 
-			return resp, err
-		})
-		defer mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, testCollectionID, testCollectionName, defaultSchema()))
+	s.sch = entity.NewSchema().WithName(testCollectionName).
+		WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
+		WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithDim(testVectorDim))
+	s.schDynamic = entity.NewSchema().WithName(testCollectionName).WithDynamicFieldEnabled(true).
+		WithField(entity.NewField().WithName("ID").WithDataType(entity.FieldTypeVarChar).WithIsPrimaryKey(true)).
+		WithField(entity.NewField().WithName("$meta").WithDataType(entity.FieldTypeJSON).WithIsDynamic(true)).
+		WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithDim(testVectorDim))
+}
 
-		// non-exist collection
-		_, err := c.Query(ctx, "non-exists-collection", []string{}, "pk in {}", []string{})
-		assert.Error(t, err)
+func (s *QuerySuite) TestQueryByPksFail() {
+	c := s.client
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	partName := "part_1"
+	idCol := entity.NewColumnInt64("ID", []int64{1})
+	s.Run("service_not_ready", func() {
+		_, err := (&GrpcClient{}).QueryByPks(ctx, testCollectionName, []string{partName}, idCol, []string{"ID"})
+		s.Error(err)
+		s.ErrorIs(err, ErrClientNotReady)
 	})
 
-	t.Run("Query Service error", func(t *testing.T) {
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		mockServer.SetInjection(MHasPartition, hasPartitionInjection(t, testCollectionName, false, partName))
-		defer mockServer.DelInjection(MHasPartition)
+	s.Run("ids_len_0", func() {
+		_, err := c.QueryByPks(ctx, testCollectionName, []string{partName}, entity.NewColumnInt64("ID", []int64{}), []string{"ID"})
+		s.Error(err)
+	})
 
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			_, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
+	s.Run("query_failed", func() {
+		defer s.resetMock()
+		s.setupDescribeCollectionError(common.ErrorCode_Success, errors.New("mock error"))
 
-			resp := &server.QueryResults{}
-			return resp, errors.New("mocked error")
-		})
-		defer mockServer.DelInjection(MQuery)
+		_, err := c.QueryByPks(ctx, testCollectionName, []string{partName}, idCol, []string{"ID"})
+		s.Error(err)
+	})
+}
 
-		_, err := c.Query(ctx, testCollectionName, []string{}, "int64 in {1}", []string{"*"})
-		assert.Error(t, err)
+func (s *QuerySuite) TestQueryFail() {
+	c := s.client
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			_, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
+	partName := "part_1"
+	s.resetMock()
 
-			resp := &server.QueryResults{}
-			resp.Status = &common.Status{
-				ErrorCode: common.ErrorCode_UnexpectedError,
-				Reason:    "mocked error",
-			}
-			return resp, nil
-		})
-
-		_, err = c.Query(ctx, testCollectionName, []string{}, "int64 in {1}", []string{"*"})
-		assert.Error(t, err)
+	s.Run("service_not_ready", func() {
+		_, err := (&GrpcClient{}).Query(ctx, testCollectionName, []string{partName}, "", []string{"ID"}, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
+		s.ErrorIs(err, ErrClientNotReady)
+	})
 
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			_, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
+	s.Run("fail_describecollection_error", func() {
+		defer s.resetMock()
 
-			resp := &server.QueryResults{}
-			s, err := SuccessStatus()
-			resp.Status = s
-			resp.FieldsData = []*schema.FieldData{
-				{
-					Type:      schema.DataType_String,
-					FieldName: "int64",
-					Field: &schema.FieldData_Scalars{
-						Scalars: &schema.ScalarField{
-							Data: &schema.ScalarField_LongData{
-								LongData: &schema.LongArray{
-									Data: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
-								},
-							},
-						},
-					},
-				},
-			}
+		s.setupDescribeCollectionError(common.ErrorCode_Success, errors.New("mock error"))
 
-			return resp, err
-		})
-		_, err = c.Query(ctx, testCollectionName, []string{}, "int64 in {1}", []string{"*"})
-		assert.Error(t, err)
+		_, err := c.Query(ctx, testCollectionName, []string{partName}, "", []string{"ID"}, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
+	})
 
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			_, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
+	s.Run("fail_describecollection_errcode", func() {
+		defer s.resetMock()
 
-			resp := &server.QueryResults{}
-			s, err := SuccessStatus()
-			resp.Status = s
-			resp.FieldsData = []*schema.FieldData{
-				{
-					Type:      schema.DataType_FloatVector,
-					FieldName: "int64",
-					Field: &schema.FieldData_Scalars{
-						Scalars: &schema.ScalarField{
-							Data: &schema.ScalarField_LongData{
-								LongData: &schema.LongArray{
-									Data: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
-								},
-							},
-						},
-					},
-				},
-			}
+		s.setupDescribeCollectionError(common.ErrorCode_UnexpectedError, nil)
 
-			return resp, err
-		})
-		_, err = c.Query(ctx, testCollectionName, []string{}, "int64 in {1}", []string{"*"})
-		assert.Error(t, err)
+		_, err := c.Query(ctx, testCollectionName, []string{partName}, "", []string{"ID"}, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
 	})
 
-}
+	s.Run("fail_guaranteed_non_custom_cl", func() {
+		defer s.resetMock()
 
-func TestGrpcQueryByPks(t *testing.T) {
-	ctx := context.Background()
-	c := testClient(ctx, t)
-	partName := "testPart"
+		s.setupDescribeCollection(testCollectionName, s.sch)
 
-	t.Run("normal query by pks", func(t *testing.T) {
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		mockServer.SetInjection(MHasPartition, hasPartitionInjection(t, testCollectionName, false, partName))
-		defer mockServer.DelInjection(MHasPartition)
+		_, err := c.Query(ctx, testCollectionName, []string{partName}, "", []string{"ID"}, WithSearchQueryConsistencyLevel(entity.ClStrong), WithGuaranteeTimestamp(1000000))
+		s.Error(err)
+	})
 
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			req, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
-			assert.Equal(t, testCollectionName, req.GetCollectionName())
-			if len(req.GetPartitionNames()) > 0 {
-				assert.Equal(t, partName, req.GetPartitionNames()[0])
-			}
+	s.Run("fail_search_error", func() {
+		defer s.resetMock()
 
-			resp := &server.QueryResults{}
-			s, err := SuccessStatus()
-			resp.Status = s
-			resp.FieldsData = []*schema.FieldData{
-				{
-					Type:      schema.DataType_Int64,
-					FieldName: "int64",
-					Field: &schema.FieldData_Scalars{
-						Scalars: &schema.ScalarField{
-							Data: &schema.ScalarField_LongData{
-								LongData: &schema.LongArray{
-									Data: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
-								},
-							},
-						},
-					},
-				},
-				{
-					Type:      schema.DataType_FloatVector,
-					FieldName: testVectorField,
-					Field: &schema.FieldData_Vectors{
-						Vectors: &schema.VectorField{
-							Dim: 1,
-							Data: &schema.VectorField_FloatVector{
-								FloatVector: &schema.FloatArray{
-									Data: []float32{0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0},
-								},
-							},
-						},
-					},
-				},
-			}
+		s.setupDescribeCollection(testCollectionName, s.sch)
+		s.mock.EXPECT().Query(mock.Anything, mock.AnythingOfType("*milvuspb.QueryRequest")).
+			Return(nil, errors.New("mock error"))
 
-			return resp, err
-		})
-		defer mockServer.DelInjection(MQuery)
+		_, err := c.Query(ctx, testCollectionName, []string{partName}, "ID in {1}", []string{"ID"}, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
+	})
 
-		columns, err := c.QueryByPks(ctx, testCollectionName, []string{partName}, entity.NewColumnInt64(testPrimaryField, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}), []string{"int64", testVectorField})
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(columns))
-		assert.Equal(t, entity.FieldTypeInt64, columns[0].Type())
-		assert.Equal(t, entity.FieldTypeFloatVector, columns[1].Type())
-		assert.Equal(t, 10, columns[0].Len())
+	s.Run("fail_search_errcode", func() {
+		defer s.resetMock()
 
-		colInt64, ok := columns[0].(*entity.ColumnInt64)
-		assert.True(t, ok)
-		assert.ElementsMatch(t, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, colInt64.Data())
+		s.setupDescribeCollection(testCollectionName, s.sch)
+		s.mock.EXPECT().Query(mock.Anything, mock.AnythingOfType("*milvuspb.QueryRequest")).
+			Return(&server.QueryResults{Status: &common.Status{ErrorCode: common.ErrorCode_UnexpectedError}}, nil)
 
-		// query with options
-		columns, err = c.QueryByPks(ctx, testCollectionName, []string{partName}, entity.NewColumnInt64(testPrimaryField, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}), []string{"int64", testVectorField}, WithSearchQueryConsistencyLevel(entity.ClSession), WithTravelTimestamp(0))
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(columns))
-		assert.Equal(t, entity.FieldTypeInt64, columns[0].Type())
-		assert.Equal(t, entity.FieldTypeFloatVector, columns[1].Type())
-		assert.Equal(t, 10, columns[0].Len())
-
-		colInt64, ok = columns[0].(*entity.ColumnInt64)
-		assert.True(t, ok)
-		assert.ElementsMatch(t, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, colInt64.Data())
+		_, err := c.Query(ctx, testCollectionName, []string{partName}, "ID in {1}", []string{"ID"}, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
 	})
 
-	t.Run("normal query by varchar pks", func(t *testing.T) {
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		mockServer.SetInjection(MHasPartition, hasPartitionInjection(t, testCollectionName, false, partName))
-		defer mockServer.DelInjection(MHasPartition)
-		mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, testCollectionID, testCollectionName, varCharSchema()))
-		defer mockServer.SetInjection(MDescribeCollection, describeCollectionInjection(t, testCollectionID, testCollectionName, defaultSchema()))
-
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			req, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
-			assert.Equal(t, testCollectionName, req.GetCollectionName())
-			if len(req.GetPartitionNames()) > 0 {
-				assert.Equal(t, partName, req.GetPartitionNames()[0])
-			}
+	s.Run("fail_response_type_error", func() {
+		defer s.resetMock()
 
-			resp := &server.QueryResults{}
-			s, err := SuccessStatus()
-			resp.Status = s
-			resp.FieldsData = []*schema.FieldData{
-				{
-					Type:      schema.DataType_VarChar,
-					FieldName: "varchar",
-					Field: &schema.FieldData_Scalars{
-						Scalars: &schema.ScalarField{
-							Data: &schema.ScalarField_StringData{
-								StringData: &schema.StringArray{
-									Data: []string{"1", "2", "3", "4", "5", "6", "7", "8", "9", "10"},
-								},
-							},
-						},
-					},
-				},
-				{
-					Type:      schema.DataType_FloatVector,
-					FieldName: testVectorField,
-					Field: &schema.FieldData_Vectors{
-						Vectors: &schema.VectorField{
-							Dim: 1,
-							Data: &schema.VectorField_FloatVector{
-								FloatVector: &schema.FloatArray{
-									Data: []float32{0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0},
+		s.setupDescribeCollection(testCollectionName, s.sch)
+		s.mock.EXPECT().Query(mock.Anything, mock.AnythingOfType("*milvuspb.QueryRequest")).
+			Return(&server.QueryResults{
+				Status: &common.Status{ErrorCode: common.ErrorCode_Success},
+				FieldsData: []*schema.FieldData{
+					{
+						FieldName: "ID",
+						Type:      schema.DataType_String, //wrong data type here
+						Field: &schema.FieldData_Scalars{
+							Scalars: &schema.ScalarField{
+								Data: &schema.ScalarField_LongData{
+									LongData: &schema.LongArray{
+										Data: []int64{1},
+									},
 								},
 							},
 						},
 					},
 				},
-			}
+			}, nil)
 
-			return resp, err
-		})
-		defer mockServer.DelInjection(MQuery)
-
-		columns, err := c.QueryByPks(ctx, testCollectionName, []string{partName}, entity.NewColumnVarChar("varchar", []string{"1", "2", "3", "4", "5", "6", "7", "8", "9", "10"}), []string{"varchar", testVectorField})
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(columns))
-		assert.Equal(t, entity.FieldTypeVarChar, columns[0].Type())
-		assert.Equal(t, entity.FieldTypeFloatVector, columns[1].Type())
-		assert.Equal(t, 10, columns[0].Len())
-
-		colPK, ok := columns[0].(*entity.ColumnVarChar)
-		assert.True(t, ok)
-		assert.ElementsMatch(t, []string{"1", "2", "3", "4", "5", "6", "7", "8", "9", "10"}, colPK.Data())
-	})
-
-	t.Run("Bad request querys", func(t *testing.T) {
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		mockServer.SetInjection(MHasPartition, hasPartitionInjection(t, testCollectionName, false, partName))
-		defer mockServer.DelInjection(MHasPartition)
-
-		// non-exist collection
-		_, err := c.QueryByPks(ctx, "non-exists-collection", []string{}, entity.NewColumnInt64("pk", []int64{}), []string{})
-		assert.Error(t, err)
-
-		// non-exist parition
-		_, err = c.QueryByPks(ctx, testCollectionName, []string{"non-exists-part"}, entity.NewColumnInt64("pk", []int64{}), []string{})
-		assert.Error(t, err)
-
-		// zero length pk
-		_, err = c.QueryByPks(ctx, testCollectionName, []string{}, entity.NewColumnInt64("pk", []int64{}), []string{})
-		assert.Error(t, err)
-
-		// string pk field
-		_, err = c.QueryByPks(ctx, testCollectionName, []string{}, entity.NewColumnString("pk", []string{"1"}), []string{})
-		assert.Error(t, err)
+		_, err := c.Query(ctx, testCollectionName, []string{partName}, "ID in {1}", []string{"ID"}, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.Error(err)
 	})
+}
 
-	t.Run("Query Service error", func(t *testing.T) {
-		mockServer.SetInjection(MHasCollection, hasCollectionDefault)
-		mockServer.SetInjection(MHasPartition, hasPartitionInjection(t, testCollectionName, false, partName))
-		defer mockServer.DelInjection(MHasPartition)
-
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			_, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
-
-			resp := &server.QueryResults{}
-			return resp, errors.New("mocked error")
-		})
-		defer mockServer.DelInjection(MQuery)
-
-		_, err := c.QueryByPks(ctx, testCollectionName, []string{}, entity.NewColumnInt64(testPrimaryField, []int64{1}), []string{"*"})
-		assert.Error(t, err)
-
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			_, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
+func (s *QuerySuite) TestQuerySuccess() {
+	c := s.client
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 
-			resp := &server.QueryResults{}
-			resp.Status = &common.Status{
-				ErrorCode: common.ErrorCode_UnexpectedError,
-				Reason:    "mocked error",
-			}
-			return resp, nil
-		})
+	partName := "part_1"
+	s.resetMock()
 
-		_, err = c.QueryByPks(ctx, testCollectionName, []string{}, entity.NewColumnInt64(testPrimaryField, []int64{1}), []string{"*"})
-		assert.Error(t, err)
+	expr := "ID in {1}"
 
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			_, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
+	s.Run("non_dynamic", func() {
+		defer s.resetMock()
 
-			resp := &server.QueryResults{}
-			s, err := SuccessStatus()
-			resp.Status = s
-			resp.FieldsData = []*schema.FieldData{
-				{
-					Type:      schema.DataType_String,
-					FieldName: "int64",
-					Field: &schema.FieldData_Scalars{
-						Scalars: &schema.ScalarField{
-							Data: &schema.ScalarField_LongData{
-								LongData: &schema.LongArray{
-									Data: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
-								},
-							},
-						},
-					},
+		s.setupDescribeCollection(testCollectionName, s.sch)
+		s.mock.EXPECT().Query(mock.Anything, mock.AnythingOfType("*milvuspb.QueryRequest")).
+			Run(func(_ context.Context, req *server.QueryRequest) {}).
+			Return(&server.QueryResults{
+				Status: getSuccessStatus(),
+				FieldsData: []*schema.FieldData{
+					s.getInt64FieldData("ID", []int64{1}),
+					s.getFloatVectorFieldData("vector", 1, []float32{0.1}),
 				},
-			}
-
-			return resp, err
-		})
-		_, err = c.QueryByPks(ctx, testCollectionName, []string{}, entity.NewColumnInt64(testPrimaryField, []int64{1}), []string{"*"})
-		assert.Error(t, err)
+			}, nil)
+
+		rs, err := c.Query(ctx, testCollectionName, []string{partName}, expr, []string{"ID", "vector"}, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.NoError(err)
+		s.Require().Equal(2, len(rs))
+		colID, ok := rs.GetColumn("ID").(*entity.ColumnInt64)
+		s.Require().True(ok)
+		s.NotNil(colID)
+		v, err := colID.Get(0)
+		s.NoError(err)
+		s.EqualValues(1, v)
+		colVector, ok := rs.GetColumn("vector").(*entity.ColumnFloatVector)
+		s.Require().True(ok)
+		s.NotNil(colVector)
+		v, err = colVector.Get(0)
+		s.NoError(err)
+		s.EqualValues([]float32{0.1}, v)
+	})
 
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			_, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
+	s.Run("dynamic_schema", func() {
+		defer s.resetMock()
 
-			resp := &server.QueryResults{}
-			s, err := SuccessStatus()
-			resp.Status = s
-			resp.FieldsData = []*schema.FieldData{
-				{
-					Type:      schema.DataType_FloatVector,
-					FieldName: "int64",
-					Field: &schema.FieldData_Scalars{
-						Scalars: &schema.ScalarField{
-							Data: &schema.ScalarField_LongData{
-								LongData: &schema.LongArray{
-									Data: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
-								},
-							},
-						},
-					},
+		s.setupDescribeCollection(testCollectionName, s.schDynamic)
+		s.mock.EXPECT().Query(mock.Anything, mock.AnythingOfType("*milvuspb.QueryRequest")).
+			Run(func(_ context.Context, req *server.QueryRequest) {}).
+			Return(&server.QueryResults{
+				Status: getSuccessStatus(),
+				FieldsData: []*schema.FieldData{
+					s.getVarcharFieldData("ID", []string{"1"}),
+					s.getFloatVectorFieldData("vector", 1, []float32{0.1}),
+					s.getJSONBytesFieldData("$meta", [][]byte{
+						[]byte(`{"A": 123, "B": "456"}`),
+						[]byte(`{"B": "abc", "A": 456}`),
+					}, true),
 				},
-			}
-
-			return resp, err
-		})
-		_, err = c.QueryByPks(ctx, testCollectionName, []string{}, entity.NewColumnInt64(testPrimaryField, []int64{1}), []string{"*"})
-		assert.Error(t, err)
+			}, nil)
+
+		rs, err := c.Query(ctx, testCollectionName, []string{partName}, `id in {"1"}`, []string{"ID", "vector", "A"}, WithSearchQueryConsistencyLevel(entity.ClStrong))
+		s.NoError(err)
+		s.Require().Equal(3, len(rs))
+		colID, ok := rs.GetColumn("ID").(*entity.ColumnVarChar)
+		s.Require().True(ok)
+		s.NotNil(colID)
+		v, err := colID.Get(0)
+		s.NoError(err)
+		s.EqualValues("1", v)
+		colVector, ok := rs.GetColumn("vector").(*entity.ColumnFloatVector)
+		s.Require().True(ok)
+		s.NotNil(colVector)
+		v, err = colVector.Get(0)
+		s.NoError(err)
+		s.EqualValues([]float32{0.1}, v)
+
+		columnA := rs.GetColumn("A").(*entity.ColumnDynamic)
+		s.Require().True(ok)
+		s.Require().NotNil(columnA)
+		v, err = columnA.GetInt64(0)
+		s.NoError(err)
+		s.Equal(int64(123), v)
 	})
 }
 
+func TestQuery(t *testing.T) {
+	suite.Run(t, new(QuerySuite))
+}
+
 func TestGrpcCalcDistanceWithIDs(t *testing.T) {
 	ctx := context.Background()
 	t.Run("bad client calls CalcDistance", func(t *testing.T) {
@@ -1174,56 +865,17 @@ func TestIsCollectionPrimaryKey(t *testing.T) {
 
 func TestEstRowSize(t *testing.T) {
 	// a schema contains all supported vector
-	sch := &entity.Schema{
-		CollectionName: testCollectionName,
-		AutoID:         false,
-		Fields: []*entity.Field{
-			{
-				Name:       testPrimaryField,
-				DataType:   entity.FieldTypeInt64,
-				PrimaryKey: true,
-				AutoID:     true,
-			},
-			{
-				Name:     "attr1",
-				DataType: entity.FieldTypeInt8,
-			},
-			{
-				Name:     "attr2",
-				DataType: entity.FieldTypeInt16,
-			},
-			{
-				Name:     "attr3",
-				DataType: entity.FieldTypeInt32,
-			},
-			{
-				Name:     "attr4",
-				DataType: entity.FieldTypeFloat,
-			},
-			{
-				Name:     "attr5",
-				DataType: entity.FieldTypeDouble,
-			},
-			{
-				Name:     "attr6",
-				DataType: entity.FieldTypeBool,
-			},
-			{
-				Name:     testVectorField,
-				DataType: entity.FieldTypeFloatVector,
-				TypeParams: map[string]string{
-					entity.TypeParamDim: fmt.Sprintf("%d", testVectorDim),
-				},
-			},
-			{
-				Name:     "binary_vector",
-				DataType: entity.FieldTypeBinaryVector,
-				TypeParams: map[string]string{
-					entity.TypeParamDim: fmt.Sprintf("%d", testVectorDim),
-				},
-			},
-		},
-	}
+	sch := entity.NewSchema().WithName(testCollectionName).WithAutoID(false).
+		WithField(entity.NewField().WithName(testPrimaryField).WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true).WithIsAutoID(true)).
+		WithField(entity.NewField().WithName("attr1").WithDataType(entity.FieldTypeInt8)).
+		WithField(entity.NewField().WithName("attr2").WithDataType(entity.FieldTypeInt16)).
+		WithField(entity.NewField().WithName("attr3").WithDataType(entity.FieldTypeInt32)).
+		WithField(entity.NewField().WithName("attr4").WithDataType(entity.FieldTypeFloat)).
+		WithField(entity.NewField().WithName("attr5").WithDataType(entity.FieldTypeDouble)).
+		WithField(entity.NewField().WithName("attr6").WithDataType(entity.FieldTypeBool)).
+		WithField(entity.NewField().WithName("attr6").WithDataType(entity.FieldTypeBool)).
+		WithField(entity.NewField().WithName(testVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(testVectorDim)).
+		WithField(entity.NewField().WithName("binary_vector").WithDataType(entity.FieldTypeBinaryVector).WithDim(testVectorDim))
 
 	// one row
 	columnID := entity.NewColumnInt64(testPrimaryField, []int64{0})
diff --git a/client/insert_test.go b/client/insert_test.go
new file mode 100644
index 0000000..dc3d02a
--- /dev/null
+++ b/client/insert_test.go
@@ -0,0 +1,274 @@
+// Copyright (C) 2019-2021 Zilliz. All rights reserved.
+//
+// Licensed 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 client
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/errors"
+	common "github.com/milvus-io/milvus-proto/go-api/commonpb"
+	server "github.com/milvus-io/milvus-proto/go-api/milvuspb"
+	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
+	"github.com/milvus-io/milvus-sdk-go/v2/entity"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/suite"
+)
+
+type InsertSuite struct {
+	MockSuiteBase
+}
+
+func (s *InsertSuite) TestInsertFail() {
+	c := s.client
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	s.Run("collection_not_exist", func() {
+		defer s.resetMock()
+		s.setupHasCollection()
+		_, err := c.Insert(ctx, testCollectionName, "")
+		s.Error(err)
+	})
+
+	s.Run("partition_not_exist", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName)
+
+		_, err := c.Insert(ctx, testCollectionName, "partition_name")
+		s.Error(err)
+	})
+
+	s.Run("field_not_exist", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, "partition_1")
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithField(entity.NewField().WithIsPrimaryKey(true).WithIsAutoID(true).WithName("ID").WithDataType(entity.FieldTypeInt64)).
+			WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		_, err := c.Insert(ctx, testCollectionName, "partition_1",
+			entity.NewColumnInt64("extra", []int64{1}),
+		)
+		s.Error(err)
+	})
+
+	s.Run("missing_field", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, "partition_1")
+
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithField(entity.NewField().WithIsPrimaryKey(true).WithIsAutoID(true).WithName("ID").WithDataType(entity.FieldTypeInt64)).
+			WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		_, err := c.Insert(ctx, testCollectionName, "partition_1",
+			entity.NewColumnInt64("ID", []int64{1}),
+		)
+		s.Error(err)
+	})
+
+	s.Run("column_len_not_match", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, "partition_1")
+
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithField(entity.NewField().WithIsPrimaryKey(true).WithIsAutoID(true).WithName("ID").WithDataType(entity.FieldTypeInt64)).
+			WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		_, err := c.Insert(ctx, testCollectionName, "partition_1",
+			entity.NewColumnInt64("ID", []int64{1, 2}),
+			entity.NewColumnFloatVector("vector", 128, [][]float32{}),
+		)
+		s.Error(err)
+	})
+
+	s.Run("duplicated column", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, "partition_1")
+
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithField(entity.NewField().WithIsPrimaryKey(true).WithIsAutoID(true).WithName("ID").WithDataType(entity.FieldTypeInt64)).
+			WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		s.mock.EXPECT().Insert(mock.Anything, mock.AnythingOfType("*milvuspb.InsertRequest")).
+			Run(func(ctx context.Context, req *server.InsertRequest) {
+				s.Equal(1, len(req.GetFieldsData()))
+			}).Return(&server.MutationResult{
+			Status: &common.Status{},
+			IDs: &schema.IDs{
+				IdField: &schema.IDs_IntId{
+					IntId: &schema.LongArray{
+						Data: []int64{1},
+					},
+				},
+			},
+		}, nil)
+
+		_, err := c.Insert(ctx, testCollectionName, "partition_1",
+			entity.NewColumnFloatVector("vector", 128, generateFloatVector(1, 128)),
+			entity.NewColumnFloatVector("vector", 128, generateFloatVector(1, 128)),
+		)
+
+		s.Error(err)
+	})
+
+	s.Run("dim_not_match", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, "partition_1")
+
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithField(entity.NewField().WithIsPrimaryKey(true).WithIsAutoID(true).WithName("ID").WithDataType(entity.FieldTypeInt64)).
+			WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		_, err := c.Insert(ctx, testCollectionName, "partition_1",
+			entity.NewColumnInt64("ID", []int64{1}),
+			entity.NewColumnFloatVector("vector", 8, [][]float32{{0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8}}),
+		)
+		s.Error(err)
+	})
+
+	s.Run("server_insert_fail", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, "partition_1")
+
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithField(entity.NewField().WithIsPrimaryKey(true).WithIsAutoID(true).WithName("ID").WithDataType(entity.FieldTypeInt64)).
+			WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		s.mock.EXPECT().Insert(mock.Anything, mock.AnythingOfType("*milvuspb.InsertRequest")).Return(
+			&server.MutationResult{Status: &common.Status{ErrorCode: common.ErrorCode_UnexpectedError}}, nil,
+		)
+
+		_, err := c.Insert(ctx, testCollectionName, "partition_1",
+			entity.NewColumnFloatVector("vector", 128, generateFloatVector(1, 128)),
+		)
+
+		s.Error(err)
+	})
+
+	s.Run("server_connection_error", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, "partition_1")
+
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithField(entity.NewField().WithIsPrimaryKey(true).WithIsAutoID(true).WithName("ID").WithDataType(entity.FieldTypeInt64)).
+			WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		s.mock.EXPECT().Insert(mock.Anything, mock.AnythingOfType("*milvuspb.InsertRequest")).Return(
+			nil, errors.New("mocked error"),
+		)
+
+		_, err := c.Insert(ctx, testCollectionName, "partition_1",
+			entity.NewColumnFloatVector("vector", 128, generateFloatVector(1, 128)),
+		)
+
+		s.Error(err)
+	})
+}
+
+func (s *InsertSuite) TestInsertSuccess() {
+	c := s.client
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	s.Run("non_dynamic_schema", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, "partition_1")
+
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithField(entity.NewField().WithIsPrimaryKey(true).WithIsAutoID(true).WithName("ID").WithDataType(entity.FieldTypeInt64)).
+			WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		s.mock.EXPECT().Insert(mock.Anything, mock.AnythingOfType("*milvuspb.InsertRequest")).
+			Run(func(ctx context.Context, req *server.InsertRequest) {
+				s.Equal(1, len(req.GetFieldsData()))
+			}).Return(&server.MutationResult{
+			Status: &common.Status{},
+			IDs: &schema.IDs{
+				IdField: &schema.IDs_IntId{
+					IntId: &schema.LongArray{
+						Data: []int64{1},
+					},
+				},
+			},
+		}, nil)
+
+		r, err := c.Insert(ctx, testCollectionName, "partition_1",
+			entity.NewColumnFloatVector("vector", 128, generateFloatVector(1, 128)),
+		)
+
+		s.NoError(err)
+		s.Equal(1, r.Len())
+	})
+
+	s.Run("dynamic_field_schema", func() {
+		defer s.resetMock()
+		s.setupHasCollection(testCollectionName)
+		s.setupHasPartition(testCollectionName, "partition_1")
+
+		s.setupDescribeCollection(testCollectionName, entity.NewSchema().
+			WithDynamicFieldEnabled(true).
+			WithField(entity.NewField().WithIsPrimaryKey(true).WithIsAutoID(true).WithName("ID").WithDataType(entity.FieldTypeInt64)).
+			WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithTypeParams(entity.TypeParamDim, "128")),
+		)
+
+		s.mock.EXPECT().Insert(mock.Anything, mock.AnythingOfType("*milvuspb.InsertRequest")).
+			Run(func(ctx context.Context, req *server.InsertRequest) {
+				s.Equal(2, len(req.GetFieldsData()))
+				var found bool
+				for _, fd := range req.GetFieldsData() {
+					if fd.GetFieldName() == "" && fd.GetIsDynamic() {
+						found = true
+						break
+					}
+				}
+				s.True(found)
+			}).Return(&server.MutationResult{
+			Status: &common.Status{},
+			IDs: &schema.IDs{
+				IdField: &schema.IDs_IntId{
+					IntId: &schema.LongArray{
+						Data: []int64{1},
+					},
+				},
+			},
+		}, nil)
+
+		r, err := c.Insert(ctx, testCollectionName, "partition_1",
+			entity.NewColumnInt64("extra", []int64{1}),
+			entity.NewColumnFloatVector("vector", 128, generateFloatVector(1, 128)),
+		)
+
+		s.NoError(err)
+		s.Equal(1, r.Len())
+	})
+}
+
+func TestGrpcInsert(t *testing.T) {
+	suite.Run(t, new(InsertSuite))
+}
diff --git a/client/row_test.go b/client/row_test.go
index fcef937..68ded83 100644
--- a/client/row_test.go
+++ b/client/row_test.go
@@ -2,17 +2,20 @@ package client
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"reflect"
 	"testing"
 
+	"github.com/cockroachdb/errors"
+
 	"github.com/golang/protobuf/proto"
 	common "github.com/milvus-io/milvus-proto/go-api/commonpb"
 	server "github.com/milvus-io/milvus-proto/go-api/milvuspb"
 	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) {
@@ -85,107 +88,232 @@ 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.setupDescribeCollectionError(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.setupDescribeCollectionError(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")),
+		)
+
+		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/columns_dynamic_test.go b/entity/columns_dynamic_test.go
new file mode 100644
index 0000000..017b665
--- /dev/null
+++ b/entity/columns_dynamic_test.go
@@ -0,0 +1,147 @@
+package entity
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+)
+
+type ColumnDynamicSuite struct {
+	suite.Suite
+}
+
+func (s *ColumnDynamicSuite) TestGetInt() {
+	cases := []struct {
+		input       string
+		expectErr   bool
+		expectValue int64
+	}{
+		{`{"field": 1000000000000000001}`, false, 1000000000000000001},
+		{`{"field": 4418489049307132905}`, false, 4418489049307132905},
+		{`{"other_field": 4418489049307132905}`, true, 0},
+		{`{"field": "string"}`, true, 0},
+	}
+
+	for _, c := range cases {
+		s.Run(c.input, func() {
+			column := NewColumnDynamic(&ColumnJSONBytes{
+				values: [][]byte{[]byte(c.input)},
+			}, "field")
+			v, err := column.GetInt64(0)
+			if c.expectErr {
+				s.Error(err)
+				return
+			}
+			s.NoError(err)
+			s.Equal(c.expectValue, v)
+		})
+	}
+}
+
+func (s *ColumnDynamicSuite) TestGetString() {
+	cases := []struct {
+		input       string
+		expectErr   bool
+		expectValue string
+	}{
+		{`{"field": "abc"}`, false, "abc"},
+		{`{"field": "test"}`, false, "test"},
+		{`{"other_field": "string"}`, true, ""},
+		{`{"field": 123}`, true, ""},
+	}
+
+	for _, c := range cases {
+		s.Run(c.input, func() {
+			column := NewColumnDynamic(&ColumnJSONBytes{
+				values: [][]byte{[]byte(c.input)},
+			}, "field")
+			v, err := column.GetString(0)
+			if c.expectErr {
+				s.Error(err)
+				return
+			}
+			s.NoError(err)
+			s.Equal(c.expectValue, v)
+		})
+	}
+}
+
+func (s *ColumnDynamicSuite) TestGetBool() {
+	cases := []struct {
+		input       string
+		expectErr   bool
+		expectValue bool
+	}{
+		{`{"field": true}`, false, true},
+		{`{"field": false}`, false, false},
+		{`{"other_field": true}`, true, false},
+		{`{"field": "test"}`, true, false},
+	}
+
+	for _, c := range cases {
+		s.Run(c.input, func() {
+			column := NewColumnDynamic(&ColumnJSONBytes{
+				values: [][]byte{[]byte(c.input)},
+			}, "field")
+			v, err := column.GetBool(0)
+			if c.expectErr {
+				s.Error(err)
+				return
+			}
+			s.NoError(err)
+			s.Equal(c.expectValue, v)
+		})
+	}
+}
+
+func (s *ColumnDynamicSuite) TestGetDouble() {
+	cases := []struct {
+		input       string
+		expectErr   bool
+		expectValue float64
+	}{
+		{`{"field": 1}`, false, 1.0},
+		{`{"field": 6231.123}`, false, 6231.123},
+		{`{"other_field": 1.0}`, true, 0},
+		{`{"field": "string"}`, true, 0},
+	}
+
+	for _, c := range cases {
+		s.Run(c.input, func() {
+			column := NewColumnDynamic(&ColumnJSONBytes{
+				values: [][]byte{[]byte(c.input)},
+			}, "field")
+			v, err := column.GetDouble(0)
+			if c.expectErr {
+				s.Error(err)
+				return
+			}
+			s.NoError(err)
+			s.Less(v-c.expectValue, 1e-10)
+		})
+	}
+
+}
+
+func (s *ColumnDynamicSuite) TestIndexOutOfRange() {
+	var err error
+	column := NewColumnDynamic(&ColumnJSONBytes{}, "field")
+
+	s.Equal("field", column.Name())
+
+	_, err = column.GetInt64(0)
+	s.Error(err)
+
+	_, err = column.GetString(0)
+	s.Error(err)
+
+	_, err = column.GetBool(0)
+	s.Error(err)
+
+	_, err = column.GetDouble(0)
+	s.Error(err)
+}
+
+func TestColumnDynamic(t *testing.T) {
+	suite.Run(t, new(ColumnDynamicSuite))
+}
diff --git a/entity/columns_json_test.go b/entity/columns_json_test.go
index 90da76e..5595dbd 100644
--- a/entity/columns_json_test.go
+++ b/entity/columns_json_test.go
@@ -22,7 +22,7 @@ func (s *ColumnJSONBytesSuite) TestAttrMethods() {
 	columnLen := 8 + rand.Intn(10)
 
 	v := make([][]byte, columnLen)
-	column := NewColumnJSONBytes(columnName, v)
+	column := NewColumnJSONBytes(columnName, v).WithIsDynamic(true)
 
 	s.Run("test_meta", func() {
 		ft := FieldTypeJSON
@@ -70,7 +70,6 @@ func (s *ColumnJSONBytesSuite) TestAttrMethods() {
 		err = column.AppendValue(1)
 		s.Error(err)
 	})
-
 }
 
 func TestColumnJSONBytes(t *testing.T) {
diff --git a/entity/columns_scalar_gen_test.go b/entity/columns_scalar_gen_test.go
index b800e7f..58b35f3 100755
--- a/entity/columns_scalar_gen_test.go
+++ b/entity/columns_scalar_gen_test.go
@@ -1,7 +1,7 @@
 // Code generated by go generate; DO NOT EDIT
-// This file is generated by go genrated at 2022-01-25 14:11:22.832059835 +0800 CST m=+0.002903820
+// This file is generated by go generate 
 
-package entity
+package entity 
 
 import (
 	"fmt"
@@ -9,8 +9,8 @@ import (
 	"testing"
 	"time"
 
-	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 	"github.com/stretchr/testify/assert"
+	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 )
 
 func TestColumnBool(t *testing.T) {
@@ -60,7 +60,7 @@ func TestFieldDataBoolColumn(t *testing.T) {
 	len := rand.Intn(10) + 8
 	name := fmt.Sprintf("fd_Bool_%d", rand.Int())
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Bool,
+		Type: schema.DataType_Bool,
 		FieldName: name,
 	}
 
@@ -77,7 +77,7 @@ func TestFieldDataBoolColumn(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, len)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+ 
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeBool, column.Type())
@@ -92,12 +92,13 @@ func TestFieldDataBoolColumn(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 
+	
 	t.Run("nil data", func(t *testing.T) {
 		fd.Field = nil
 		_, err := FieldDataColumn(fd, 0, len)
 		assert.NotNil(t, err)
 	})
-
+	
 	t.Run("get all data", func(t *testing.T) {
 		fd.Field = &schema.FieldData_Scalars{
 			Scalars: &schema.ScalarField{
@@ -111,7 +112,7 @@ func TestFieldDataBoolColumn(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, -1)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+		
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeBool, column.Type())
@@ -165,7 +166,7 @@ func TestFieldDataInt8Column(t *testing.T) {
 	len := rand.Intn(10) + 8
 	name := fmt.Sprintf("fd_Int8_%d", rand.Int())
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Int8,
+		Type: schema.DataType_Int8,
 		FieldName: name,
 	}
 
@@ -182,7 +183,7 @@ func TestFieldDataInt8Column(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, len)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+ 
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeInt8, column.Type())
@@ -197,12 +198,13 @@ func TestFieldDataInt8Column(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 
+	
 	t.Run("nil data", func(t *testing.T) {
 		fd.Field = nil
 		_, err := FieldDataColumn(fd, 0, len)
 		assert.NotNil(t, err)
 	})
-
+	
 	t.Run("get all data", func(t *testing.T) {
 		fd.Field = &schema.FieldData_Scalars{
 			Scalars: &schema.ScalarField{
@@ -216,7 +218,7 @@ func TestFieldDataInt8Column(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, -1)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+		
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeInt8, column.Type())
@@ -270,7 +272,7 @@ func TestFieldDataInt16Column(t *testing.T) {
 	len := rand.Intn(10) + 8
 	name := fmt.Sprintf("fd_Int16_%d", rand.Int())
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Int16,
+		Type: schema.DataType_Int16,
 		FieldName: name,
 	}
 
@@ -287,7 +289,7 @@ func TestFieldDataInt16Column(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, len)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+ 
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeInt16, column.Type())
@@ -302,12 +304,13 @@ func TestFieldDataInt16Column(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 
+	
 	t.Run("nil data", func(t *testing.T) {
 		fd.Field = nil
 		_, err := FieldDataColumn(fd, 0, len)
 		assert.NotNil(t, err)
 	})
-
+	
 	t.Run("get all data", func(t *testing.T) {
 		fd.Field = &schema.FieldData_Scalars{
 			Scalars: &schema.ScalarField{
@@ -321,7 +324,7 @@ func TestFieldDataInt16Column(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, -1)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+		
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeInt16, column.Type())
@@ -375,7 +378,7 @@ func TestFieldDataInt32Column(t *testing.T) {
 	len := rand.Intn(10) + 8
 	name := fmt.Sprintf("fd_Int32_%d", rand.Int())
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Int32,
+		Type: schema.DataType_Int32,
 		FieldName: name,
 	}
 
@@ -392,7 +395,7 @@ func TestFieldDataInt32Column(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, len)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+ 
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeInt32, column.Type())
@@ -407,12 +410,13 @@ func TestFieldDataInt32Column(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 
+	
 	t.Run("nil data", func(t *testing.T) {
 		fd.Field = nil
 		_, err := FieldDataColumn(fd, 0, len)
 		assert.NotNil(t, err)
 	})
-
+	
 	t.Run("get all data", func(t *testing.T) {
 		fd.Field = &schema.FieldData_Scalars{
 			Scalars: &schema.ScalarField{
@@ -426,7 +430,7 @@ func TestFieldDataInt32Column(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, -1)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+		
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeInt32, column.Type())
@@ -480,7 +484,7 @@ func TestFieldDataInt64Column(t *testing.T) {
 	len := rand.Intn(10) + 8
 	name := fmt.Sprintf("fd_Int64_%d", rand.Int())
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Int64,
+		Type: schema.DataType_Int64,
 		FieldName: name,
 	}
 
@@ -497,7 +501,7 @@ func TestFieldDataInt64Column(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, len)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+ 
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeInt64, column.Type())
@@ -512,12 +516,13 @@ func TestFieldDataInt64Column(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 
+	
 	t.Run("nil data", func(t *testing.T) {
 		fd.Field = nil
 		_, err := FieldDataColumn(fd, 0, len)
 		assert.NotNil(t, err)
 	})
-
+	
 	t.Run("get all data", func(t *testing.T) {
 		fd.Field = &schema.FieldData_Scalars{
 			Scalars: &schema.ScalarField{
@@ -531,7 +536,7 @@ func TestFieldDataInt64Column(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, -1)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+		
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeInt64, column.Type())
@@ -585,7 +590,7 @@ func TestFieldDataFloatColumn(t *testing.T) {
 	len := rand.Intn(10) + 8
 	name := fmt.Sprintf("fd_Float_%d", rand.Int())
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Float,
+		Type: schema.DataType_Float,
 		FieldName: name,
 	}
 
@@ -602,7 +607,7 @@ func TestFieldDataFloatColumn(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, len)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+ 
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeFloat, column.Type())
@@ -617,12 +622,13 @@ func TestFieldDataFloatColumn(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 
+	
 	t.Run("nil data", func(t *testing.T) {
 		fd.Field = nil
 		_, err := FieldDataColumn(fd, 0, len)
 		assert.NotNil(t, err)
 	})
-
+	
 	t.Run("get all data", func(t *testing.T) {
 		fd.Field = &schema.FieldData_Scalars{
 			Scalars: &schema.ScalarField{
@@ -636,7 +642,7 @@ func TestFieldDataFloatColumn(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, -1)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+		
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeFloat, column.Type())
@@ -690,7 +696,7 @@ func TestFieldDataDoubleColumn(t *testing.T) {
 	len := rand.Intn(10) + 8
 	name := fmt.Sprintf("fd_Double_%d", rand.Int())
 	fd := &schema.FieldData{
-		Type:      schema.DataType_Double,
+		Type: schema.DataType_Double,
 		FieldName: name,
 	}
 
@@ -707,7 +713,7 @@ func TestFieldDataDoubleColumn(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, len)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+ 
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeDouble, column.Type())
@@ -722,12 +728,13 @@ func TestFieldDataDoubleColumn(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 
+	
 	t.Run("nil data", func(t *testing.T) {
 		fd.Field = nil
 		_, err := FieldDataColumn(fd, 0, len)
 		assert.NotNil(t, err)
 	})
-
+	
 	t.Run("get all data", func(t *testing.T) {
 		fd.Field = &schema.FieldData_Scalars{
 			Scalars: &schema.ScalarField{
@@ -741,7 +748,7 @@ func TestFieldDataDoubleColumn(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, -1)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+		
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeDouble, column.Type())
@@ -795,7 +802,7 @@ func TestFieldDataStringColumn(t *testing.T) {
 	len := rand.Intn(10) + 8
 	name := fmt.Sprintf("fd_String_%d", rand.Int())
 	fd := &schema.FieldData{
-		Type:      schema.DataType_String,
+		Type: schema.DataType_String,
 		FieldName: name,
 	}
 
@@ -812,7 +819,7 @@ func TestFieldDataStringColumn(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, len)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+ 
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeString, column.Type())
@@ -827,12 +834,13 @@ func TestFieldDataStringColumn(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 
+	
 	t.Run("nil data", func(t *testing.T) {
 		fd.Field = nil
 		_, err := FieldDataColumn(fd, 0, len)
 		assert.NotNil(t, err)
 	})
-
+	
 	t.Run("get all data", func(t *testing.T) {
 		fd.Field = &schema.FieldData_Scalars{
 			Scalars: &schema.ScalarField{
@@ -846,9 +854,10 @@ func TestFieldDataStringColumn(t *testing.T) {
 		column, err := FieldDataColumn(fd, 0, -1)
 		assert.Nil(t, err)
 		assert.NotNil(t, column)
-
+		
 		assert.Equal(t, name, column.Name())
 		assert.Equal(t, len, column.Len())
 		assert.Equal(t, FieldTypeString, column.Type())
 	})
 }
+
diff --git a/entity/columns_vector_gen_test.go b/entity/columns_vector_gen_test.go
index e3548bf..1aea399 100755
--- a/entity/columns_vector_gen_test.go
+++ b/entity/columns_vector_gen_test.go
@@ -1,7 +1,7 @@
 // Code generated by go generate; DO NOT EDIT
-// This file is generated by go genrated at 2022-01-26 15:50:36.315715425 +0800 CST m=+0.003539472
+// This file is generated by go generate 
 
-package entity
+package entity 
 
 import (
 	"fmt"
@@ -19,16 +19,16 @@ func TestColumnBinaryVector(t *testing.T) {
 	columnLen := 12 + rand.Intn(10)
 	dim := ([]int{64, 128, 256, 512})[rand.Intn(4)]
 
-	v := make([][]byte, 0, columnLen)
+	v := make([][]byte,0, columnLen)
 	dlen := dim
 	dlen /= 8
-
+	
 	for i := 0; i < columnLen; i++ {
 		entry := make([]byte, dlen)
 		v = append(v, entry)
 	}
 	column := NewColumnBinaryVector(columnName, dim, v)
-
+	
 	t.Run("test meta", func(t *testing.T) {
 		ft := FieldTypeBinaryVector
 		assert.Equal(t, "BinaryVector", ft.Name())
@@ -43,13 +43,13 @@ func TestColumnBinaryVector(t *testing.T) {
 		assert.Equal(t, FieldTypeBinaryVector, column.Type())
 		assert.Equal(t, columnLen, column.Len())
 		assert.Equal(t, dim, column.Dim())
-		assert.Equal(t, v, column.Data())
-
+		assert.Equal(t ,v, column.Data())
+		
 		var ev []byte
 		err := column.AppendValue(ev)
 		assert.Equal(t, columnLen+1, column.Len())
 		assert.Nil(t, err)
-
+		
 		err = column.AppendValue(struct{}{})
 		assert.Equal(t, columnLen+1, column.Len())
 		assert.NotNil(t, err)
@@ -70,7 +70,7 @@ func TestColumnBinaryVector(t *testing.T) {
 			Type:      schema.DataType_BinaryVector,
 			FieldName: columnName,
 		}
-		_, err := FieldDataVector(fd)
+		_, err := FieldDataVector(fd) 
 		assert.Error(t, err)
 	})
 
@@ -82,15 +82,16 @@ func TestColumnFloatVector(t *testing.T) {
 	columnLen := 12 + rand.Intn(10)
 	dim := ([]int{64, 128, 256, 512})[rand.Intn(4)]
 
-	v := make([][]float32, 0, columnLen)
+	v := make([][]float32,0, columnLen)
 	dlen := dim
-
+	
+	
 	for i := 0; i < columnLen; i++ {
 		entry := make([]float32, dlen)
 		v = append(v, entry)
 	}
 	column := NewColumnFloatVector(columnName, dim, v)
-
+	
 	t.Run("test meta", func(t *testing.T) {
 		ft := FieldTypeFloatVector
 		assert.Equal(t, "FloatVector", ft.Name())
@@ -105,13 +106,13 @@ func TestColumnFloatVector(t *testing.T) {
 		assert.Equal(t, FieldTypeFloatVector, column.Type())
 		assert.Equal(t, columnLen, column.Len())
 		assert.Equal(t, dim, column.Dim())
-		assert.Equal(t, v, column.Data())
-
+		assert.Equal(t ,v, column.Data())
+		
 		var ev []float32
 		err := column.AppendValue(ev)
 		assert.Equal(t, columnLen+1, column.Len())
 		assert.Nil(t, err)
-
+		
 		err = column.AppendValue(struct{}{})
 		assert.Equal(t, columnLen+1, column.Len())
 		assert.NotNil(t, err)
@@ -132,8 +133,9 @@ func TestColumnFloatVector(t *testing.T) {
 			Type:      schema.DataType_FloatVector,
 			FieldName: columnName,
 		}
-		_, err := FieldDataVector(fd)
+		_, err := FieldDataVector(fd) 
 		assert.Error(t, err)
 	})
 
 }
+
diff --git a/entity/rows_test.go b/entity/rows_test.go
index a46db96..ea8e48c 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,30 @@ 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().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.NoError(err)
+	})
+}
+
+func TestRows(t *testing.T) {
+	suite.Run(t, new(RowsSuite))
+}
diff --git a/entity/schema_test.go b/entity/schema_test.go
index a488e31..f79f2da 100644
--- a/entity/schema_test.go
+++ b/entity/schema_test.go
@@ -5,6 +5,7 @@ import (
 
 	common "github.com/milvus-io/milvus-proto/go-api/commonpb"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/suite"
 )
 
 func TestCL_CommonCL(t *testing.T) {
@@ -21,22 +22,9 @@ func TestCL_CommonCL(t *testing.T) {
 
 func TestFieldSchema(t *testing.T) {
 	fields := []*Field{
-		{
-			Name:        "int_field",
-			DataType:    FieldTypeInt64,
-			AutoID:      true,
-			PrimaryKey:  true,
-			Description: "int_field desc",
-			TypeParams:  map[string]string{},
-		},
-		{
-			Name:        "string_field",
-			DataType:    FieldTypeString,
-			AutoID:      false,
-			PrimaryKey:  true,
-			Description: "false_field desc",
-			TypeParams:  map[string]string{"max_len": "32"}, // not applied, just testing value
-		},
+		NewField().WithName("int_field").WithDataType(FieldTypeInt64).WithIsAutoID(true).WithIsPrimaryKey(true).WithDescription("int_field desc"),
+		NewField().WithName("string_field").WithDataType(FieldTypeString).WithIsAutoID(false).WithIsPrimaryKey(true).WithIsDynamic(false).WithTypeParams("max_len", "32").WithDescription("string_field desc"),
+		NewField().WithName("partition_key").WithDataType(FieldTypeInt32).WithIsPartitionKey(true),
 	}
 
 	for _, field := range fields {
@@ -46,6 +34,8 @@ func TestFieldSchema(t *testing.T) {
 		assert.EqualValues(t, field.DataType, fieldSchema.GetDataType())
 		assert.Equal(t, field.AutoID, fieldSchema.GetAutoID())
 		assert.Equal(t, field.PrimaryKey, fieldSchema.GetIsPrimaryKey())
+		assert.Equal(t, field.IsPartitionKey, fieldSchema.GetIsPartitionKey())
+		assert.Equal(t, field.IsDynamic, fieldSchema.GetIsDynamic())
 		assert.Equal(t, field.Description, fieldSchema.GetDescription())
 		assert.Equal(t, field.TypeParams, KvPairsMap(fieldSchema.GetTypeParams()))
 		// marshal & unmarshal, still equals
@@ -57,29 +47,66 @@ func TestFieldSchema(t *testing.T) {
 		assert.Equal(t, field.AutoID, nf.AutoID)
 		assert.Equal(t, field.PrimaryKey, nf.PrimaryKey)
 		assert.Equal(t, field.Description, nf.Description)
-		assert.Equal(t, field.TypeParams, nf.TypeParams)
+		assert.Equal(t, field.IsDynamic, nf.IsDynamic)
+		assert.Equal(t, field.IsPartitionKey, nf.IsPartitionKey)
+		assert.EqualValues(t, field.TypeParams, nf.TypeParams)
 	}
+
+	assert.NotPanics(t, func() {
+		(&Field{}).WithTypeParams("a", "b")
+	})
 }
 
-func TestSchema(t *testing.T) {
-	schemas := []*Schema{
+type SchemaSuite struct {
+	suite.Suite
+}
+
+func (s *SchemaSuite) TestBasic() {
+	cases := []struct {
+		tag    string
+		input  *Schema
+		pkName string
+	}{
+		{
+			"test_collection",
+			NewSchema().WithName("test_collection_1").WithDescription("test_collection_1 desc").WithAutoID(false).
+				WithField(NewField().WithName("ID").WithDataType(FieldTypeInt64).WithIsPrimaryKey(true)).
+				WithField(NewField().WithName("vector").WithDataType(FieldTypeFloatVector).WithDim(128)),
+			"ID",
+		},
 		{
-			CollectionName: "test_collection_1",
-			AutoID:         false,
-			Description:    "test_collection_1 decription",
+			"dynamic_schema",
+			NewSchema().WithName("dynamic_schema").WithDescription("dynamic_schema desc").WithAutoID(true).WithDynamicFieldEnabled(true).
+				WithField(NewField().WithName("ID").WithDataType(FieldTypeVarChar).WithMaxLength(256)).
+				WithField(NewField().WithName("$meta").WithIsDynamic(true)),
+			"",
 		},
 	}
-	for _, sch := range schemas {
-		p := sch.ProtoMessage()
-		assert.Equal(t, sch.CollectionName, p.GetName())
-		assert.Equal(t, sch.AutoID, p.GetAutoID())
-		assert.Equal(t, sch.Description, p.GetDescription())
 
-		nsch := &Schema{}
-		nsch = nsch.ReadProto(p)
+	for _, c := range cases {
+		s.Run(c.tag, func() {
+			sch := c.input
+			p := sch.ProtoMessage()
+			s.Equal(sch.CollectionName, p.GetName())
+			s.Equal(sch.AutoID, p.GetAutoID())
+			s.Equal(sch.Description, p.GetDescription())
+			s.Equal(sch.EnableDynamicField, p.GetEnableDynamicField())
+			s.Equal(len(sch.Fields), len(p.GetFields()))
+
+			nsch := &Schema{}
+			nsch = nsch.ReadProto(p)
 
-		assert.Equal(t, sch.CollectionName, nsch.CollectionName)
-		assert.Equal(t, sch.AutoID, nsch.AutoID)
-		assert.Equal(t, sch.Description, nsch.Description)
+			s.Equal(sch.CollectionName, nsch.CollectionName)
+			s.Equal(sch.AutoID, nsch.AutoID)
+			s.Equal(sch.Description, nsch.Description)
+			s.Equal(sch.EnableDynamicField, nsch.EnableDynamicField)
+			s.Equal(len(sch.Fields), len(nsch.Fields))
+			s.Equal(c.pkName, sch.PKFieldName())
+			s.Equal(c.pkName, nsch.PKFieldName())
+		})
 	}
 }
+
+func TestSchema(t *testing.T) {
+	suite.Run(t, new(SchemaSuite))
+}
diff --git a/test/testcases/insert_test.go b/test/testcases/insert_test.go
index 9cc9a0c..87fa9e7 100644
--- a/test/testcases/insert_test.go
+++ b/test/testcases/insert_test.go
@@ -226,7 +226,7 @@ func TestInsertColumnsMismatchFields(t *testing.T) {
 
 	// len(column) > len(fields)
 	_, errInsert2 := mc.Insert(ctx, collName, "", intColumn, floatColumn, vecColumn, floatColumn)
-	common.CheckErr(t, errInsert2, false, "len(columns) mismatch the len(fields), len(columns): 4, len(fields): 3")
+	common.CheckErr(t, errInsert2, false, "duplicated column")
 
 	// order(column) != order(fields)
 	_, errInsert3 := mc.Insert(ctx, collName, "", floatColumn, vecColumn, intColumn)