Repository: milvus-io/milvus-sdk-go
Pull request #457 (2023-05-21 12:51:19)

Support search & query with dynamic schema
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/457#" 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/457?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) Report
> Merging [#457](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/457?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) (e0168ce) into [master](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/commit/5f94bce118bdbe7de8303b8da90496fb4d43de74?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) (5f94bce) will **decrease** coverage by `0.75%`.
> The diff coverage is `66.66%`.

[![Impacted file tree graph](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/457/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/457?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     #457      +/-   ##
==========================================
- Coverage   86.28%   85.54%   -0.75%     
==========================================
  Files          33       34       +1     
  Lines        3114     3224     +110     
==========================================
+ Hits         2687     2758      +71     
- Misses        332      371      +39     
  Partials       95       95              
```


| [Impacted Files](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/457?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/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L2NvbHVtbnMuZ28=) | `73.03% <0.00%> (-15.41%)` | :arrow_down: |
| [entity/schema.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L3NjaGVtYS5nbw==) | `93.93% <60.00%> (-0.94%)` | :arrow_down: |
| [client/client.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L2NsaWVudC5nbw==) | `75.38% <66.66%> (-0.43%)` | :arrow_down: |
| [client/data.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-Y2xpZW50L2RhdGEuZ28=) | `72.70% <83.33%> (+2.05%)` | :arrow_up: |
| [entity/columns\_dynamic.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/457?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 file with indirect coverage changes](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/457/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io)

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

This pull-request has been approved by: *<a href="https://github.com/milvus-io/milvus-sdk-go/pull/457#" title="Author self-approved">congqixia</a>*, *<a href="https://github.com/milvus-io/milvus-sdk-go/pull/457#pullrequestreview-1435674733" 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.go b/client/client.go
index 72ebfda..8ef72b7 100644
--- a/client/client.go
+++ b/client/client.go
@@ -125,9 +125,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,
@@ -198,11 +198,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     //[]entity.Column // 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
 }
 
 var DefaultGrpcOpts = []grpc.DialOption{
diff --git a/client/data.go b/client/data.go
index e503fa1..a67420f 100644
--- a/client/data.go
+++ b/client/data.go
@@ -39,10 +39,18 @@ func (c *GrpcClient) Search(ctx context.Context, collName string, partitions []s
 	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
@@ -76,21 +84,55 @@ 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 (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 dynamicField != nil && fieldData.GetFieldName() == dynamicField.Name {
+			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 dynamicField == nil {
+				return nil, errors.New("output fields not match when dynamic field disabled")
+			}
+			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()
@@ -111,7 +153,7 @@ func PKs2Expr(backName string, ids entity.Column) string {
 }
 
 // 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
 	}
@@ -129,17 +171,21 @@ func (c *GrpcClient) QueryByPks(ctx context.Context, collectionName string, part
 }
 
 // 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...)
@@ -176,22 +222,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
+	// query always has pk field as output
+	hasPK := false
+	pkName := sch.PKFieldName()
+	for _, output := range outputFields {
+		if output == pkName {
+			hasPK = true
+			break
 		}
-		column, err := entity.FieldDataColumn(fieldData, 0, -1)
-		if err != nil {
-			return nil, err
-		}
-		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
diff --git a/entity/columns.go b/entity/columns.go
index e636f2a..2d7d102 100644
--- a/entity/columns.go
+++ b/entity/columns.go
@@ -237,6 +237,49 @@ func FieldDataColumn(fd *schema.FieldData, begin, end int) (Column, error) {
 			return NewColumnJSONBytes(fd.GetFieldName(), data.JsonData.GetData()[begin:]), nil
 		}
 		return NewColumnJSONBytes(fd.GetFieldName(), data.JsonData.GetData()[begin:end]), 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/schema.go b/entity/schema.go
index 270db1d..ca56457 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"
 )
@@ -52,15 +54,18 @@ type Schema struct {
 	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
@@ -82,6 +87,19 @@ func (s *Schema) WithField(f *Field) *Schema {
 	return s
 }
 
+// GetDynamicField returns dynamic field when EnabledDynamicField is enabled.
+func (s *Schema) GetDynamicField() *Field {
+	if !s.EnableDynamicField {
+		return nil
+	}
+	for _, field := range s.Fields {
+		if field.IsDynamic {
+			return field
+		}
+	}
+	return nil
+}
+
 // ProtoMessage returns corresponding server.CollectionSchema
 func (s *Schema) ProtoMessage() *schema.CollectionSchema {
 	r := &schema.CollectionSchema{
@@ -104,12 +122,22 @@ 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
@@ -138,6 +166,7 @@ func (f *Field) ProtoMessage() *schema.FieldSchema {
 	}
 }
 
+// NewField creates a new Field with map initialized.
 func NewField() *Field {
 	return &Field{
 		TypeParams:  make(map[string]string),
@@ -183,6 +212,14 @@ func (f *Field) WithTypeParams(key string, value string) *Field {
 	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
+}
+
 // ReadProto parses FieldSchema
 func (f *Field) ReadProto(p *schema.FieldSchema) *Field {
 	f.ID = p.GetFieldID()
diff --git a/go.mod b/go.mod
index 4a66d5b..73b4065 100644
--- a/go.mod
+++ b/go.mod
@@ -24,6 +24,9 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rogpeppe/go-internal v1.8.1 // indirect
 	github.com/stretchr/objx v0.5.0 // indirect
+	github.com/tidwall/gjson v1.14.4 // indirect
+	github.com/tidwall/match v1.1.1 // indirect
+	github.com/tidwall/pretty v1.2.0 // indirect
 	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd // indirect
 	golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f // indirect
 	golang.org/x/text v0.3.8 // indirect
diff --git a/go.sum b/go.sum
index 397d3f5..3af253b 100644
--- a/go.sum
+++ b/go.sum
@@ -219,6 +219,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 74834dc..f43c90c 100644
--- a/client/client_mock_test.go
+++ b/client/client_mock_test.go
@@ -12,6 +12,7 @@ import (
 	"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"
+	schema "github.com/milvus-io/milvus-proto/go-api/schemapb"
 	"github.com/milvus-io/milvus-sdk-go/v2/entity"
 	"github.com/milvus-io/milvus-sdk-go/v2/mocks"
 	"github.com/stretchr/testify/mock"
@@ -133,13 +134,78 @@ func (s *MockSuiteBase) setupDescribeCollection(collName string, schema *entity.
 	}, nil)
 }
 
-func (s *MockSuiteBase) setupDescirbeCollectionError(errorCode common.ErrorCode, err error) {
+func (s *MockSuiteBase) setupDescribeCollectionError(errorCode common.ErrorCode, err error) {
 	s.mock.EXPECT().DescribeCollection(mock.Anything, mock.AnythingOfType("*milvuspb.DescribeCollectionRequest")).
 		Return(&milvuspb.DescribeCollectionResponse{
 			Status: &common.Status{ErrorCode: errorCode},
 		}, err)
 }
 
+func (s *MockSuiteBase) getInt64FieldData(name string, data []int64) *schema.FieldData {
+	return &schema.FieldData{
+		Type:      schema.DataType_Int64,
+		FieldName: name,
+		Field: &schema.FieldData_Scalars{
+			Scalars: &schema.ScalarField{
+				Data: &schema.ScalarField_LongData{
+					LongData: &schema.LongArray{
+						Data: data,
+					},
+				},
+			},
+		},
+	}
+}
+
+func (s *MockSuiteBase) getVarcharFieldData(name string, data []string) *schema.FieldData {
+	return &schema.FieldData{
+		Type:      schema.DataType_VarChar,
+		FieldName: name,
+		Field: &schema.FieldData_Scalars{
+			Scalars: &schema.ScalarField{
+				Data: &schema.ScalarField_StringData{
+					StringData: &schema.StringArray{
+						Data: data,
+					},
+				},
+			},
+		},
+	}
+}
+
+func (s *MockSuiteBase) getJSONBytesFieldData(name string, data [][]byte) *schema.FieldData {
+	return &schema.FieldData{
+		Type:      schema.DataType_JSON,
+		FieldName: name,
+		Field: &schema.FieldData_Scalars{
+			Scalars: &schema.ScalarField{
+				Data: &schema.ScalarField_JsonData{
+					JsonData: &schema.JSONArray{
+						Data: data,
+					},
+				},
+			},
+		},
+	}
+}
+
+func (s *MockSuiteBase) getFloatVectorFieldData(name string, dim int64, data []float32) *schema.FieldData {
+	return &schema.FieldData{
+		Type:      schema.DataType_FloatVector,
+		FieldName: name,
+		Field: &schema.FieldData_Vectors{
+			Vectors: &schema.VectorField{
+				Dim: dim,
+				Data: &schema.VectorField_FloatVector{
+					FloatVector: &schema.FloatArray{
+						Data: data,
+					},
+				},
+			},
+		},
+	}
+}
+
 // ref https://stackoverflow.com/questions/42102496/testing-a-grpc-service
 
 var (
@@ -832,6 +898,10 @@ func (m *MockServer) CheckHealth(ctx context.Context, req *server.CheckHealthReq
 	return &server.CheckHealthResponse{Status: s}, err
 }
 
+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 41f8887..efcaf7d 100644
--- a/client/data_test.go
+++ b/client/data_test.go
@@ -21,13 +21,14 @@ import (
 	"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 TestGrpcClientFlush(t *testing.T) {
@@ -267,666 +268,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("$meta", [][]byte{
+							[]byte(`{"A": 123, "B": "456"}`),
+							[]byte(`{"B": "abc", "A": 456}`),
+						}),
 					},
-				},
-				{
-					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)
-
-		mockServer.SetInjection(MQuery, func(_ context.Context, raw proto.Message) (proto.Message, error) {
-			_, ok := raw.(*server.QueryRequest)
-			if !ok {
-				t.FailNow()
-			}
+	s.Run("ids_len_0", func() {
+		_, err := c.QueryByPks(ctx, testCollectionName, []string{partName}, entity.NewColumnInt64("ID", []int64{}), []string{"ID"})
+		s.Error(err)
+	})
 
-			resp := &server.QueryResults{}
-			return resp, errors.New("mockServer.d error")
-		})
-		defer mockServer.DelInjection(MQuery)
+	s.Run("query_failed", func() {
+		defer s.resetMock()
+		s.setupDescribeCollectionError(common.ErrorCode_Success, errors.New("mock error"))
 
-		_, err := c.Query(ctx, testCollectionName, []string{}, "int64 in {1}", []string{"*"})
-		assert.Error(t, err)
+		_, err := c.QueryByPks(ctx, testCollectionName, []string{partName}, idCol, []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()
-			}
+func (s *QuerySuite) TestQueryFail() {
+	c := s.client
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 
-			resp := &server.QueryResults{}
-			resp.Status = &common.Status{
-				ErrorCode: common.ErrorCode_UnexpectedError,
-				Reason:    "mockServer.d error",
-			}
-			return resp, nil
-		})
+	partName := "part_1"
+	s.resetMock()
 
-		_, 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)
+func (s *QuerySuite) TestQuerySuccess() {
+	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{}
-			return resp, errors.New("mockServer.d error")
-		})
-		defer mockServer.DelInjection(MQuery)
+	expr := "ID in {1}"
 
-		_, 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()
-			}
+	s.Run("non_dynamic", func() {
+		defer s.resetMock()
 
-			resp := &server.QueryResults{}
-			resp.Status = &common.Status{
-				ErrorCode: common.ErrorCode_UnexpectedError,
-				Reason:    "mockServer.d error",
-			}
-			return resp, nil
-		})
-
-		_, 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()
-			}
-
-			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}`),
+					}),
 				},
-			}
-
-			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) {
diff --git a/client/row_test.go b/client/row_test.go
index 26a423a..877315b 100644
--- a/client/row_test.go
+++ b/client/row_test.go
@@ -155,7 +155,7 @@ func (s *InsertByRowsSuite) TestFails() {
 		defer s.resetMock()
 		s.setupHasCollection(testCollectionName)
 		s.setupHasPartition(testCollectionName, partName)
-		s.setupDescirbeCollectionError(common.ErrorCode_Success, errors.New("mock error"))
+		s.setupDescribeCollectionError(common.ErrorCode_Success, errors.New("mock error"))
 		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
 		s.Error(err)
 	})
@@ -164,7 +164,7 @@ func (s *InsertByRowsSuite) TestFails() {
 		defer s.resetMock()
 		s.setupHasCollection(testCollectionName)
 		s.setupHasPartition(testCollectionName, partName)
-		s.setupDescirbeCollectionError(common.ErrorCode_UnexpectedError, nil)
+		s.setupDescribeCollectionError(common.ErrorCode_UnexpectedError, nil)
 		_, err := c.InsertByRows(ctx, testCollectionName, partName, []entity.Row{entity.RowBase{}})
 		s.Error(err)
 	})
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/schema_test.go b/entity/schema_test.go
index daea0bd..c4efd9c 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) {
@@ -51,24 +52,90 @@ func TestFieldSchema(t *testing.T) {
 	})
 }
 
-func TestSchema(t *testing.T) {
-	schemas := []*Schema{
-		NewSchema().WithName("test_collection_1").WithDescription("test_collection_1 desc").WithAutoID(false),
-		NewSchema().WithName("dynamic_schema").WithDescription("dynamic_schema desc").WithAutoID(true).WithDynamicFieldEnabled(true),
+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)),
+			"ID",
+		},
+		{
+			"dynamic_schema",
+			NewSchema().WithName("dynamic_schema").WithDescription("dynamic_schema desc").WithAutoID(true).WithDynamicFieldEnabled(true).
+				WithField(NewField().WithName("$meta").WithIsDynamic(true)),
+			"",
+		},
+	}
+
+	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)
+
+			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())
+		})
 	}
-	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)
+func (s *SchemaSuite) TestGetDynamicField() {
+	testCases := []struct {
+		tag         string
+		input       *Schema
+		expectValue bool
+	}{
+		{
+			"dynamic_field_disabled",
+			NewSchema().WithField(NewField().WithIsDynamic(true)),
+			false,
+		},
+		{
+			"dynamic_enabled_not_found",
+			NewSchema().WithDynamicFieldEnabled(true).WithField(NewField().WithName("$meta").WithIsDynamic(false)),
+			false,
+		},
+		{
+			"dynamic_enabled_found",
+			NewSchema().WithDynamicFieldEnabled(true).WithField(NewField().WithName("$meta").WithIsDynamic(true)),
+			true,
+		},
+	}
 
-		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)
+	for _, c := range testCases {
+		s.Run(c.tag, func() {
+			f := c.input.GetDynamicField()
+			if c.expectValue {
+				s.Require().NotNil(f)
+				s.True(f.IsDynamic)
+			} else {
+				s.Nil(f)
+			}
+		})
 	}
 }
+
+func TestSchema(t *testing.T) {
+	suite.Run(t, new(SchemaSuite))
+}