Repository: milvus-io/milvus-sdk-go
Pull request #455 (2023-05-18 15:55:06)

Adapt dynamic field and support insert column when dynamic field enabled
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/455#" title="Author self-approved">congqixia</a>*

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

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

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

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

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

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

```diff
@@            Coverage Diff             @@
##           master     #455      +/-   ##
==========================================
- Coverage   87.30%   85.60%   -1.71%     
==========================================
  Files          32       33       +1     
  Lines        2939     3050     +111     
==========================================
+ Hits         2566     2611      +45     
- Misses        282      347      +65     
- Partials       91       92       +1     
```


| [Impacted Files](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) | Coverage Δ | |
|---|---|---|
| [client/data.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L2RhdGEuZ28=) | `70.64% <ø> (ø)` | |
| [client/row.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L3Jvdy5nbw==) | `83.48% <ø> (ø)` | |
| [entity/columns.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L2NvbHVtbnMuZ28=) | `88.43% <0.00%> (-1.22%)` | :arrow_down: |
| [entity/columns\_json.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L2NvbHVtbnNfanNvbi5nbw==) | `89.28% <0.00%> (-10.72%)` | :arrow_down: |
| [entity/columns\_scalar\_gen.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L2NvbHVtbnNfc2NhbGFyX2dlbi5nbw==) | `87.50% <0.00%> (-12.50%)` | :arrow_down: |
| [entity/columns\_varchar.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?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: |
| [entity/columns\_vector\_gen.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L2NvbHVtbnNfdmVjdG9yX2dlbi5nbw==) | `89.65% <0.00%> (-10.35%)` | :arrow_down: |
| [entity/schema.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L3NjaGVtYS5nbw==) | `81.94% <32.00%> (-11.50%)` | :arrow_down: |
| [client/insert.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/455?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L2luc2VydC5nbw==) | `79.64% <79.64%> (ø)` | |

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/455#" title="Author self-approved">congqixia</a>*, *<a href="https://github.com/milvus-io/milvus-sdk-go/pull/455#pullrequestreview-1433800615" title="LGTM">yah01</a>*

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

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

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

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

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

Patch:
diff --git a/client/client_grpc_data.go b/client/data.go
similarity index 55%
rename from client/client_grpc_data.go
rename to client/data.go
index 8d8a941..e503fa1 100644
--- a/client/client_grpc_data.go
+++ b/client/data.go
@@ -18,11 +18,9 @@ import (
 	"log"
 	"strconv"
 	"strings"
-	"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"
@@ -35,294 +33,6 @@ 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 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
-}
-
-// Upsert Index into collection with column-based format
-// collName is the collection name
-// partitionName is the partition to upsert, if not specified(empty), default partition will be used
-// columns are slice of the column-based data
-func (c *GrpcClient) Upsert(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 upsert request
-	req := &server.UpsertRequest{
-		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.Upsert(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)
-}
-
 // 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) {
@@ -724,39 +434,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
@@ -825,91 +502,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..b9ce824
--- /dev/null
+++ b/client/insert.go
@@ -0,0 +1,514 @@
+// 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
+	}
+
+	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
+	var dynamicField *entity.Field
+	if isDynamic {
+		for _, field := range colSchema.Fields {
+			if field.IsDynamic {
+				dynamicField = field
+				break
+			}
+		}
+	}
+
+	// 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 {
+		col, err := c.mergeDynamicColumns(dynamicField.Name, 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,
+					},
+				},
+			},
+		},
+	}, 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
+}
+
+// Upsert Index into collection with column-based format
+// collName is the collection name
+// partitionName is the partition to upsert, if not specified(empty), default partition will be used
+// columns are slice of the column-based data
+func (c *GrpcClient) Upsert(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 upsert request
+	req := &server.UpsertRequest{
+		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.Upsert(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)
+}
+
+// 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/client_grpc_row.go b/client/row.go
similarity index 99%
rename from client/client_grpc_row.go
rename to client/row.go
index def4276..b1d3563 100644
--- a/client/client_grpc_row.go
+++ b/client/row.go
@@ -86,6 +86,7 @@ func (c *GrpcClient) InsertByRows(ctx context.Context, collName string, partitio
 		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..4dd204b 100644
--- a/codecov.yml
+++ b/codecov.yml
@@ -38,3 +38,5 @@ ignore:
   - "**/*.pb.go"
   - "examples/*"
   - "tests/*"
+  - "mocks/*"
+  - "*_gen.go"
diff --git a/entity/columns.go b/entity/columns.go
index ab9e466..e636f2a 100644
--- a/entity/columns.go
+++ b/entity/columns.go
@@ -13,15 +13,28 @@ package entity
 
 import (
 	"encoding/binary"
-	"errors"
 	"fmt"
 	"math"
 
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
+
+	"github.com/cockroachdb/errors"
 )
 
+var ErrDataTypeNotMatch = errors.New("data type not matched")
+
 //go:generate go run gen/gen.go
 
+type ColumnBase struct{}
+
+func (b ColumnBase) GetInt64() (int64, error) {
+	return 0, ErrDataTypeNotMatch
+}
+
+func (b ColumnBase) GetFloat64() (float64, error) {
+	return 0, ErrDataTypeNotMatch
+}
+
 // Column interface field type for column-based data frame
 type Column interface {
 	Name() string
@@ -29,6 +42,7 @@ type Column interface {
 	Len() int
 	FieldData() *schema.FieldData
 	AppendValue(interface{}) error
+	Get(int) (interface{}, error)
 }
 
 // Vector interface vector used int search
diff --git a/entity/columns_json.go b/entity/columns_json.go
index 7cdbb67..5f3f9f2 100644
--- a/entity/columns_json.go
+++ b/entity/columns_json.go
@@ -31,6 +31,14 @@ func (c *ColumnJSONBytes) Len() int {
 	return len(c.values)
 }
 
+// Get returns value at index as interface{}.
+func (c *ColumnJSONBytes) Get(idx int) (interface{}, error) {
+	if idx < 0 || idx > c.Len() {
+		return nil, errors.New("index out of range")
+	}
+	return c.values[idx], nil
+}
+
 // FieldData return column data mapped to schema.FieldData.
 func (c *ColumnJSONBytes) FieldData() *schema.FieldData {
 	fd := &schema.FieldData{
diff --git a/entity/columns_scalar_gen.go b/entity/columns_scalar_gen.go
index d13e22a..48625c1 100755
--- a/entity/columns_scalar_gen.go
+++ b/entity/columns_scalar_gen.go
@@ -31,6 +31,15 @@ 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{
@@ -108,6 +117,15 @@ 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{
@@ -185,6 +203,15 @@ 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{
@@ -262,6 +289,15 @@ 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{
@@ -339,6 +375,15 @@ 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{
@@ -416,6 +461,15 @@ 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{
@@ -493,6 +547,15 @@ 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{
@@ -570,6 +633,15 @@ 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{
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 fe95760..daa2ff5 100755
--- a/entity/columns_vector_gen.go
+++ b/entity/columns_vector_gen.go
@@ -6,6 +6,8 @@ import (
 	"fmt"
 
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
+
+	"github.com/cockroachdb/errors"
 )
 
 
@@ -36,6 +38,14 @@ 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 {
 	v, ok := i.([]byte)
@@ -114,6 +124,14 @@ 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 {
 	v, ok := i.([]float32)
diff --git a/entity/gen/gen.go b/entity/gen/gen.go
index f8ea47c..2478dc8 100644
--- a/entity/gen/gen.go
+++ b/entity/gen/gen.go
@@ -56,6 +56,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{
@@ -121,6 +130,8 @@ import (
 	"fmt"
 
 	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
+
+	"github.com/cockroachdb/errors"
 )
 
 {{ range .Types }}{{with.}}
@@ -151,6 +162,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}})
diff --git a/entity/schema.go b/entity/schema.go
index ca304e5..270db1d 100644
--- a/entity/schema.go
+++ b/entity/schema.go
@@ -45,18 +45,50 @@ 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
+}
+
+func NewSchema() *Schema {
+	return &Schema{}
+}
+
+func (s *Schema) WithName(name string) *Schema {
+	s.CollectionName = name
+	return s
+}
+
+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 {
@@ -74,6 +106,7 @@ func (s *Schema) ReadProto(p *schema.CollectionSchema) *Schema {
 	for _, fp := range p.GetFields() {
 		s.Fields = append(s.Fields, (&Field{}).ReadProto(fp))
 	}
+	s.EnableDynamicField = p.GetEnableDynamicField()
 	return s
 }
 
@@ -87,6 +120,7 @@ type Field struct {
 	DataType    FieldType
 	TypeParams  map[string]string
 	IndexParams map[string]string
+	IsDynamic   bool
 }
 
 // ProtoMessage generates corresponding FieldSchema
@@ -100,7 +134,53 @@ func (f *Field) ProtoMessage() *schema.FieldSchema {
 		DataType:     schema.DataType(f.DataType),
 		TypeParams:   MapKvPairs(f.TypeParams),
 		IndexParams:  MapKvPairs(f.IndexParams),
+		IsDynamic:    f.IsDynamic,
+	}
+}
+
+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) WithTypeParams(key string, value string) *Field {
+	if f.TypeParams == nil {
+		f.TypeParams = make(map[string]string)
 	}
+	f.TypeParams[key] = value
+	return f
 }
 
 // ReadProto parses FieldSchema
@@ -113,6 +193,7 @@ 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()
 
 	return f
 }
diff --git a/go.mod b/go.mod
index c3b8d32..4a66d5b 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.17
 require (
 	github.com/golang/protobuf v1.5.2
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
-	github.com/milvus-io/milvus-proto/go-api v0.0.0-20230418025426-3735aafecc6c
+	github.com/milvus-io/milvus-proto/go-api v0.0.0-20230517025117-8ba62a3f3a63
 	github.com/stretchr/testify v1.8.1
 	google.golang.org/grpc v1.48.0
 	google.golang.org/grpc/examples v0.0.0-20220617181431-3e7b97febc7f
diff --git a/go.sum b/go.sum
index 0d4a902..397d3f5 100644
--- a/go.sum
+++ b/go.sum
@@ -154,12 +154,15 @@ github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27k
 github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw=
 github.com/mediocregopher/radix/v3 v3.4.2/go.mod h1:8FL3F6UQRXHXIBSPUs5h0RybMF8i4n7wVopoX3x7Bv8=
 github.com/microcosm-cc/bluemonday v1.0.2/go.mod h1:iVP4YcDBq+n/5fb23BhYFvIMq/leAFZyRl6bYmGDlGc=
+github.com/milvus-io/milvus-proto v2.2.8+incompatible/go.mod h1:urRUo4TMLWgew3AYu4Ak1oYr7KcV6AO2momWggKrDIw=
 github.com/milvus-io/milvus-proto/go-api v0.0.0-20230105121931-9f9303dcc729 h1:hsb1ifdNe3qlXi1YY5dWPPzWMNZmnqe5uunYPYK3gd0=
 github.com/milvus-io/milvus-proto/go-api v0.0.0-20230105121931-9f9303dcc729/go.mod h1:148qnlmZ0Fdm1Fq+Mj/OW2uDoEP25g3mjh0vMGtkgmk=
 github.com/milvus-io/milvus-proto/go-api v0.0.0-20230303054144-16f081962572 h1:QOimLfT1VwjH9jUkq2SN3uT/WGfgcD4pYBoFOGUMwjM=
 github.com/milvus-io/milvus-proto/go-api v0.0.0-20230303054144-16f081962572/go.mod h1:148qnlmZ0Fdm1Fq+Mj/OW2uDoEP25g3mjh0vMGtkgmk=
 github.com/milvus-io/milvus-proto/go-api v0.0.0-20230418025426-3735aafecc6c h1:9BaqfI2UoA2OJY89ogizV+b6vDlcwTguZT3rwUJl7yY=
 github.com/milvus-io/milvus-proto/go-api v0.0.0-20230418025426-3735aafecc6c/go.mod h1:148qnlmZ0Fdm1Fq+Mj/OW2uDoEP25g3mjh0vMGtkgmk=
+github.com/milvus-io/milvus-proto/go-api v0.0.0-20230517025117-8ba62a3f3a63 h1:V962mSHjOFUbuMgAXziBdbYPOCVZmN1MkqEeKpME+MA=
+github.com/milvus-io/milvus-proto/go-api v0.0.0-20230517025117-8ba62a3f3a63/go.mod h1:148qnlmZ0Fdm1Fq+Mj/OW2uDoEP25g3mjh0vMGtkgmk=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
diff --git a/mocks/MilvusServiceServer.go b/mocks/MilvusServiceServer.go
index b7c338a..26cc023 100644
--- a/mocks/MilvusServiceServer.go
+++ b/mocks/MilvusServiceServer.go
@@ -215,6 +215,53 @@ func (_c *MilvusServiceServer_CheckHealth_Call) Return(_a0 *milvuspb.CheckHealth
 	return _c
 }
 
+// Connect provides a mock function with given fields: _a0, _a1
+func (_m *MilvusServiceServer) Connect(_a0 context.Context, _a1 *milvuspb.ConnectRequest) (*milvuspb.ConnectResponse, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *milvuspb.ConnectResponse
+	if rf, ok := ret.Get(0).(func(context.Context, *milvuspb.ConnectRequest) *milvuspb.ConnectResponse); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*milvuspb.ConnectResponse)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *milvuspb.ConnectRequest) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// MilvusServiceServer_Connect_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Connect'
+type MilvusServiceServer_Connect_Call struct {
+	*mock.Call
+}
+
+// Connect is a helper method to define mock.On call
+//  - _a0 context.Context
+//  - _a1 *milvuspb.ConnectRequest
+func (_e *MilvusServiceServer_Expecter) Connect(_a0 interface{}, _a1 interface{}) *MilvusServiceServer_Connect_Call {
+	return &MilvusServiceServer_Connect_Call{Call: _e.mock.On("Connect", _a0, _a1)}
+}
+
+func (_c *MilvusServiceServer_Connect_Call) Run(run func(_a0 context.Context, _a1 *milvuspb.ConnectRequest)) *MilvusServiceServer_Connect_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(*milvuspb.ConnectRequest))
+	})
+	return _c
+}
+
+func (_c *MilvusServiceServer_Connect_Call) Return(_a0 *milvuspb.ConnectResponse, _a1 error) *MilvusServiceServer_Connect_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
 // CreateAlias provides a mock function with given fields: _a0, _a1
 func (_m *MilvusServiceServer) CreateAlias(_a0 context.Context, _a1 *milvuspb.CreateAliasRequest) (*commonpb.Status, error) {
 	ret := _m.Called(_a0, _a1)
@@ -1719,6 +1766,53 @@ func (_c *MilvusServiceServer_GetIndexState_Call) Return(_a0 *milvuspb.GetIndexS
 	return _c
 }
 
+// GetIndexStatistics provides a mock function with given fields: _a0, _a1
+func (_m *MilvusServiceServer) GetIndexStatistics(_a0 context.Context, _a1 *milvuspb.GetIndexStatisticsRequest) (*milvuspb.GetIndexStatisticsResponse, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *milvuspb.GetIndexStatisticsResponse
+	if rf, ok := ret.Get(0).(func(context.Context, *milvuspb.GetIndexStatisticsRequest) *milvuspb.GetIndexStatisticsResponse); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*milvuspb.GetIndexStatisticsResponse)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *milvuspb.GetIndexStatisticsRequest) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// MilvusServiceServer_GetIndexStatistics_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetIndexStatistics'
+type MilvusServiceServer_GetIndexStatistics_Call struct {
+	*mock.Call
+}
+
+// GetIndexStatistics is a helper method to define mock.On call
+//  - _a0 context.Context
+//  - _a1 *milvuspb.GetIndexStatisticsRequest
+func (_e *MilvusServiceServer_Expecter) GetIndexStatistics(_a0 interface{}, _a1 interface{}) *MilvusServiceServer_GetIndexStatistics_Call {
+	return &MilvusServiceServer_GetIndexStatistics_Call{Call: _e.mock.On("GetIndexStatistics", _a0, _a1)}
+}
+
+func (_c *MilvusServiceServer_GetIndexStatistics_Call) Run(run func(_a0 context.Context, _a1 *milvuspb.GetIndexStatisticsRequest)) *MilvusServiceServer_GetIndexStatistics_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(*milvuspb.GetIndexStatisticsRequest))
+	})
+	return _c
+}
+
+func (_c *MilvusServiceServer_GetIndexStatistics_Call) Return(_a0 *milvuspb.GetIndexStatisticsResponse, _a1 error) *MilvusServiceServer_GetIndexStatistics_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
 // GetLoadState provides a mock function with given fields: _a0, _a1
 func (_m *MilvusServiceServer) GetLoadState(_a0 context.Context, _a1 *milvuspb.GetLoadStateRequest) (*milvuspb.GetLoadStateResponse, error) {
 	ret := _m.Called(_a0, _a1)


Test patch:
diff --git a/client/client_grpc_partition_test.go b/client/client_grpc_partition_test.go
index a82a575..88fd422 100644
--- a/client/client_grpc_partition_test.go
+++ b/client/client_grpc_partition_test.go
@@ -347,6 +347,7 @@ func TestGrpcClientReleasePartitions(t *testing.T) {
 
 		return SuccessStatus()
 	})
+	defer mockServer.SetInjection(MHasPartition, hasPartitionInjection(t, testCollectionName, false, "testPart"))
 
 	assert.Nil(t, c.ReleasePartitions(ctx, testCollectionName, parts))
 }
diff --git a/client/client_mock_test.go b/client/client_mock_test.go
index d1dd726..bcb5ee4 100644
--- a/client/client_mock_test.go
+++ b/client/client_mock_test.go
@@ -10,7 +10,9 @@ import (
 	"github.com/golang/protobuf/proto"
 	common "github.com/milvus-io/milvus-proto/go-api/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/federpb"
+	"github.com/milvus-io/milvus-proto/go-api/milvuspb"
 	server "github.com/milvus-io/milvus-proto/go-api/milvuspb"
+	"github.com/milvus-io/milvus-sdk-go/v2/entity"
 	"github.com/milvus-io/milvus-sdk-go/v2/mocks"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
@@ -76,17 +78,46 @@ func (s *MockSuiteBase) resetMock() {
 	}
 }
 
-func (s *MockSuiteBase) setupHasCollection(collName string) {
+func (s *MockSuiteBase) setupHasCollection(collNames ...string) {
 	s.mock.EXPECT().HasCollection(mock.Anything, mock.AnythingOfType("*milvuspb.HasCollectionRequest")).
 		Call.Return(func(ctx context.Context, req *server.HasCollectionRequest) *server.BoolResponse {
 		resp := &server.BoolResponse{Status: &common.Status{}}
+		for _, collName := range collNames {
+			if req.GetCollectionName() == collName {
+				resp.Value = true
+				break
+			}
+		}
+		return resp
+	}, nil)
+}
+
+func (s *MockSuiteBase) setupHasPartition(collName string, partNames ...string) {
+	s.mock.EXPECT().HasPartition(mock.Anything, mock.AnythingOfType("*milvuspb.HasPartitionRequest")).
+		Call.Return(func(ctx context.Context, req *server.HasPartitionRequest) *server.BoolResponse {
+		resp := &server.BoolResponse{Status: &common.Status{}}
 		if req.GetCollectionName() == collName {
-			resp.Value = true
+			for _, partName := range partNames {
+				if req.GetPartitionName() == partName {
+					resp.Value = true
+					break
+				}
+			}
 		}
 		return resp
 	}, nil)
 }
 
+func (s *MockSuiteBase) setupDescribeCollection(collName string, schema *entity.Schema) {
+	s.mock.EXPECT().DescribeCollection(mock.Anything, mock.AnythingOfType("*milvuspb.DescribeCollectionRequest")).
+		Call.Return(func(ctx context.Context, req *server.DescribeCollectionRequest) *server.DescribeCollectionResponse {
+		return &milvuspb.DescribeCollectionResponse{
+			Status: &common.Status{ErrorCode: common.ErrorCode_Success},
+			Schema: schema.ProtoMessage(),
+		}
+	}, nil)
+}
+
 // ref https://stackoverflow.com/questions/42102496/testing-a-grpc-service
 
 var (
@@ -737,6 +768,14 @@ func (m *MockServer) DescribeSegmentIndexData(_ context.Context, _ *federpb.Desc
 	panic("not implemented") // TODO: Implement
 }
 
+func (m *MockServer) GetIndexStatistics(_ context.Context, _ *server.GetIndexStatisticsRequest) (*server.GetIndexStatisticsResponse, error) {
+	panic("not implemented") // TODO: Implement
+}
+
+func (m *MockServer) Connect(_ context.Context, _ *server.ConnectRequest) (*server.ConnectResponse, error) {
+	panic("not implemented") // TODO: Implement
+}
+
 //func (m *MockServer) DescribePartition(ctx context.Context, req *server.DescribePartitionRequest) (*server.DescribePartitionResponse, error) {
 //	panic("not implemented") // TODO: Implement
 //}
diff --git a/client/client_grpc_data_test.go b/client/data_test.go
similarity index 93%
rename from client/client_grpc_data_test.go
rename to client/data_test.go
index dfa8654..41f8887 100644
--- a/client/client_grpc_data_test.go
+++ b/client/data_test.go
@@ -1,3 +1,14 @@
+// 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 (
@@ -19,96 +30,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-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)
@@ -346,7 +268,6 @@ func TestGrpcDeleteByPks(t *testing.T) {
 }
 
 func TestGrpcSearch(t *testing.T) {
-
 	ctx := context.Background()
 
 	c := testClient(ctx, t)
diff --git a/client/insert_test.go b/client/insert_test.go
new file mode 100644
index 0000000..aeed816
--- /dev/null
+++ b/client/insert_test.go
@@ -0,0 +1,275 @@
+// 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"
+
+	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"
+	"google.golang.org/grpc"
+)
+
+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, grpc.ErrClientConnClosing,
+		)
+
+		_, 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")).
+			WithField(entity.NewField().WithName("$meta").WithDataType(entity.FieldTypeJSON).WithIsDynamic(true)),
+		)
+
+		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() == "$meta" {
+						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/entity/schema_test.go b/entity/schema_test.go
index a488e31..daea0bd 100644
--- a/entity/schema_test.go
+++ b/entity/schema_test.go
@@ -21,22 +21,8 @@ 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"),
 	}
 
 	for _, field := range fields {
@@ -57,23 +43,25 @@ 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.EqualValues(t, field.TypeParams, nf.TypeParams)
 	}
+
+	assert.NotPanics(t, func() {
+		(&Field{}).WithTypeParams("a", "b")
+	})
 }
 
 func TestSchema(t *testing.T) {
 	schemas := []*Schema{
-		{
-			CollectionName: "test_collection_1",
-			AutoID:         false,
-			Description:    "test_collection_1 decription",
-		},
+		NewSchema().WithName("test_collection_1").WithDescription("test_collection_1 desc").WithAutoID(false),
+		NewSchema().WithName("dynamic_schema").WithDescription("dynamic_schema desc").WithAutoID(true).WithDynamicFieldEnabled(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())
+		assert.Equal(t, sch.EnableDynamicField, p.GetEnableDynamicField())
 
 		nsch := &Schema{}
 		nsch = nsch.ReadProto(p)
@@ -81,5 +69,6 @@ func TestSchema(t *testing.T) {
 		assert.Equal(t, sch.CollectionName, nsch.CollectionName)
 		assert.Equal(t, sch.AutoID, nsch.AutoID)
 		assert.Equal(t, sch.Description, nsch.Description)
+		assert.Equal(t, sch.EnableDynamicField, nsch.EnableDynamicField)
 	}
 }
diff --git a/test/testcases/insert_test.go b/test/testcases/insert_test.go
index a2b4366..8d24337 100644
--- a/test/testcases/insert_test.go
+++ b/test/testcases/insert_test.go
@@ -75,7 +75,8 @@ func TestInsertAutoIdPkData(t *testing.T) {
 	// insert
 	pkColumn, floatColumn, vecColumn := common.GenDefaultColumnData(0, common.DefaultNb, common.DefaultDim)
 	_, errInsert := mc.Insert(ctx, collName, "", pkColumn, floatColumn, vecColumn)
-	common.CheckErr(t, errInsert, false, "can not assign primary field data when auto id enabled")
+	//TODO change to check error code
+	common.CheckErr(t, errInsert, false, "invalid parameter") //, "can not assign primary field data when auto id enabled")
 
 	// flush and check row count
 	errFlush := mc.Flush(ctx, collName, false)
@@ -226,7 +227,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)