Repository: milvus-io/milvus-sdk-go
Pull request #466 (2023-05-25 07:17:52)

Support row interface with map[string]any
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/466#" 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/466?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) Report
> Merging [#466](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/466?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) (88e9fe1) into [master](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/commit/6368d50ab395de1e7a4291c9426e7776c202bf38?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) (6368d50) will **decrease** coverage by `0.09%`.
> The diff coverage is `78.57%`.

[![Impacted file tree graph](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/466/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/466?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     #466      +/-   ##
==========================================
- Coverage   84.85%   84.77%   -0.09%     
==========================================
  Files          35       35              
  Lines        3269     3283      +14     
==========================================
+ Hits         2774     2783       +9     
- Misses        393      397       +4     
- Partials      102      103       +1     
```


| [Impacted Files](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/466?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io) | Coverage Δ | |
|---|---|---|
| [entity/rows.go](https://app.codecov.io/gh/milvus-io/milvus-sdk-go/pull/466?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=milvus-io#diff-ZW50aXR5L3Jvd3MuZ28=) | `86.77% <78.57%> (-1.39%)` | :arrow_down: |

yah01: /lgtm

Patch:
diff --git a/entity/rows.go b/entity/rows.go
index 4c7d1ca..244a271 100644
--- a/entity/rows.go
+++ b/entity/rows.go
@@ -54,6 +54,21 @@ type Row interface {
 	Description() string
 }
 
+// MapRow is the alias type for map[string]interface{} implementing `Row` inteface with empty methods.
+type MapRow map[string]interface{}
+
+func (mr MapRow) Collection() string {
+	return ""
+}
+
+func (mr MapRow) Partition() string {
+	return ""
+}
+
+func (mr MapRow) Description() string {
+	return ""
+}
+
 // RowBase row base, returns default collection, partition name which is empty string
 type RowBase struct{}
 
@@ -321,49 +336,9 @@ func RowsToColumns(rows []Row, schemas ...*Schema) ([]Column, error) {
 	for _, row := range rows {
 		// collection schema name need not to be same, since receiver could has other names
 		v := reflect.ValueOf(row)
-		if v.Kind() == reflect.Ptr {
-			v = v.Elem()
-		}
-
-		type fieldCandi struct {
-			name    string
-			v       reflect.Value
-			options map[string]string
-		}
-		set := make(map[string]fieldCandi)
-		for i := 0; i < v.NumField(); i++ {
-			ft := v.Type().Field(i)
-			name := ft.Name
-			tag, ok := ft.Tag.Lookup(MilvusTag)
-
-			settings := make(map[string]string)
-			if ok {
-				if tag == MilvusSkipTagValue {
-					continue
-				}
-				settings = ParseTagSetting(tag, MilvusTagSep)
-				fn, has := settings[MilvusTagName]
-				if has {
-					// overwrite column to tag name
-					name = fn
-				}
-			}
-			_, ok = set[name]
-			// duplicated
-			if ok {
-				return nil, fmt.Errorf("column has duplicated name: %s when parsing field: %s", name, ft.Name)
-			}
-
-			v := v.Field(i)
-			if v.Kind() == reflect.Array {
-				v = v.Slice(0, v.Len()-1)
-			}
-
-			set[name] = fieldCandi{
-				name:    name,
-				v:       v,
-				options: settings,
-			}
+		set, err := reflectValueCandi(v)
+		if err != nil {
+			return nil, err
 		}
 
 		for idx, field := range sch.Fields {
@@ -407,3 +382,68 @@ func RowsToColumns(rows []Row, schemas ...*Schema) ([]Column, error) {
 	}
 	return columns, nil
 }
+
+type fieldCandi struct {
+	name    string
+	v       reflect.Value
+	options map[string]string
+}
+
+func reflectValueCandi(v reflect.Value) (map[string]fieldCandi, error) {
+	if v.Kind() == reflect.Ptr {
+		v = v.Elem()
+	}
+
+	result := make(map[string]fieldCandi)
+	switch v.Kind() {
+	case reflect.Map: // map[string]interface{}
+		iter := v.MapRange()
+		for iter.Next() {
+			key := iter.Key().String()
+			result[key] = fieldCandi{
+				name: key,
+				v:    iter.Value(),
+			}
+		}
+		return result, nil
+	case reflect.Struct:
+		for i := 0; i < v.NumField(); i++ {
+			ft := v.Type().Field(i)
+			name := ft.Name
+			tag, ok := ft.Tag.Lookup(MilvusTag)
+
+			settings := make(map[string]string)
+			if ok {
+				if tag == MilvusSkipTagValue {
+					continue
+				}
+				settings = ParseTagSetting(tag, MilvusTagSep)
+				fn, has := settings[MilvusTagName]
+				if has {
+					// overwrite column to tag name
+					name = fn
+				}
+			}
+			_, ok = result[name]
+			// duplicated
+			if ok {
+				return nil, fmt.Errorf("column has duplicated name: %s when parsing field: %s", name, ft.Name)
+			}
+
+			v := v.Field(i)
+			if v.Kind() == reflect.Array {
+				v = v.Slice(0, v.Len()-1)
+			}
+
+			result[name] = fieldCandi{
+				name:    name,
+				v:       v,
+				options: settings,
+			}
+		}
+
+		return result, nil
+	default:
+		return nil, fmt.Errorf("unsupport row type: %s", v.Kind().String())
+	}
+}
diff --git a/examples/dynamic_schema/dynamic_schema.go b/examples/dynamic_schema/dynamic_schema.go
new file mode 100644
index 0000000..667fedb
--- /dev/null
+++ b/examples/dynamic_schema/dynamic_schema.go
@@ -0,0 +1,285 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"math/rand"
+	"time"
+
+	"github.com/milvus-io/milvus-sdk-go/v2/client"
+	"github.com/milvus-io/milvus-sdk-go/v2/entity"
+)
+
+const (
+	milvusAddr     = `127.0.0.1:19530`
+	nEntities, dim = 3000, 128
+	collectionName = "dynamic_example"
+
+	msgFmt                                             = "\n==== %s ====\n"
+	idCol, typeCol, randomCol, sourceCol, embeddingCol = "ID", "type", "random", "source", "embeddings"
+	topK                                               = 4
+)
+
+func main() {
+	flag.Parse()
+	ctx := context.Background()
+
+	fmt.Printf(msgFmt, "start connecting to Milvus")
+	c, err := client.NewGrpcClient(ctx, milvusAddr)
+	if err != nil {
+		log.Fatalf("failed to connect to milvus, err: %v", err)
+	}
+	defer c.Close()
+
+	version, err := c.GetVersion(ctx)
+	if err != nil {
+		log.Fatal("failed to get version of Milvus server", err.Error())
+	}
+	fmt.Println("Milvus Version:", version)
+
+	// delete collection if exists
+	has, err := c.HasCollection(ctx, collectionName)
+	if err != nil {
+		log.Fatalf("failed to check collection exists, err: %v", err)
+	}
+	if has {
+		c.DropCollection(ctx, collectionName)
+	}
+
+	// create collection
+	fmt.Printf(msgFmt, "create collection `dynamic_example")
+	schema := entity.NewSchema().
+		WithName(collectionName).
+		WithDescription("dynamic schema example collection").
+		WithAutoID(false).
+		WithDynamicFieldEnabled(true).
+		WithField(entity.NewField().WithName(idCol).WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
+		WithField(entity.NewField().WithName(embeddingCol).WithDataType(entity.FieldTypeFloatVector).WithDim(dim))
+
+	if err := c.CreateCollection(ctx, schema, entity.DefaultShardNumber); err != nil { // use default shard number
+		log.Fatalf("create collection failed, err: %v", err)
+	}
+
+	// describe collection
+	fmt.Printf(msgFmt, "describe collection `dynamic_example`")
+	coll, err := c.DescribeCollection(ctx, collectionName)
+	if err != nil {
+		log.Fatal("failed to describe collection:", err.Error())
+	}
+
+	fmt.Printf("Collection %s\tDescription: %s\tDynamicEnabled: %t\n", coll.Schema.CollectionName, coll.Schema.CollectionName, coll.Schema.EnableDynamicField)
+	for _, field := range coll.Schema.Fields {
+		fmt.Printf("Field: %s\tDataType: %s\tIsDynamic: %t\n", field.Name, field.DataType.String(), field.IsDynamic)
+	}
+
+	// insert data
+	fmt.Printf(msgFmt, "start inserting with extra columns")
+	idList, randomList := make([]int64, 0, nEntities), make([]float64, 0, nEntities)
+	typeList := make([]int32, 0, nEntities)
+	embeddingList := make([][]float32, 0, nEntities)
+
+	rand.Seed(time.Now().UnixNano())
+	// generate data
+	for i := 0; i < nEntities; i++ {
+		idList = append(idList, int64(i))
+		typeList = append(typeList, int32(i%3))
+	}
+	for i := 0; i < nEntities; i++ {
+		randomList = append(randomList, rand.Float64())
+	}
+	for i := 0; i < nEntities; i++ {
+		vec := make([]float32, 0, dim)
+		for j := 0; j < dim; j++ {
+			vec = append(vec, rand.Float32())
+		}
+		embeddingList = append(embeddingList, vec)
+	}
+
+	idColData := entity.NewColumnInt64(idCol, idList)
+	randomColData := entity.NewColumnDouble(randomCol, randomList)
+	typeColData := entity.NewColumnInt32(typeCol, typeList)
+	embeddingColData := entity.NewColumnFloatVector(embeddingCol, dim, embeddingList)
+	if _, err := c.Insert(ctx, collectionName, "", idColData, randomColData, typeColData, embeddingColData); err != nil {
+		log.Fatalf("failed to insert random data into `dynamic_example, err: %v", err)
+	}
+
+	if err := c.Flush(ctx, collectionName, false); err != nil {
+		log.Fatalf("failed to flush data, err: %v", err)
+	}
+	fmt.Printf(msgFmt, "start inserting with rows")
+
+	// insert by struct
+	type DynamicRow struct {
+		entity.RowBase
+		ID     int64     `milvus:"name:ID;primary_key"`
+		Vector []float32 `milvus:"name:embeddings;dim:128"`
+		Source int32     `milvus:"name:source"`
+		Value  float64   `milvus:"name:random"`
+	}
+
+	rows := make([]entity.Row, 0, nEntities)
+	for i := 0; i < nEntities; i++ {
+		vec := make([]float32, 0, dim)
+		for j := 0; j < dim; j++ {
+			vec = append(vec, rand.Float32())
+		}
+
+		rows = append(rows, &DynamicRow{
+			ID:     int64(nEntities + i),
+			Vector: vec,
+			Source: 1,
+			Value:  rand.Float64(),
+		})
+	}
+
+	fmt.Printf(msgFmt, "start to insert by rows")
+	_, err = c.InsertByRows(ctx, collectionName, "", rows)
+	if err != nil {
+		log.Fatal("failed to insert by rows: ", err.Error())
+	}
+
+	// insert by map[string]interface{}
+	m := make(map[string]interface{})
+	m["ID"] = int64(nEntities)
+	vec := make([]float32, 0, dim)
+	for j := 0; j < dim; j++ {
+		vec = append(vec, rand.Float32())
+	}
+	m["embeddings"] = vec
+	m["source"] = int32(1)
+	m["random"] = rand.Float64()
+
+	_, err = c.InsertByRows(ctx, collectionName, "", []entity.Row{entity.MapRow(m)})
+	if err != nil {
+		log.Fatal("failed to insert by rows: ", err.Error())
+	}
+
+	if err := c.Flush(ctx, collectionName, false); err != nil {
+		log.Fatalf("failed to flush data, err: %v", err)
+	}
+
+	// build index
+	fmt.Printf(msgFmt, "start creating index IVF_FLAT")
+	idx, err := entity.NewIndexIvfFlat(entity.L2, 128)
+	if err != nil {
+		log.Fatalf("failed to create ivf flat index, err: %v", err)
+	}
+	if err := c.CreateIndex(ctx, collectionName, embeddingCol, idx, false); err != nil {
+		log.Fatalf("failed to create index, err: %v", err)
+	}
+
+	fmt.Printf(msgFmt, "start loading collection")
+	start := time.Now()
+	err = c.LoadCollection(ctx, collectionName, false)
+	if err != nil {
+		log.Fatalf("failed to load collection, err: %v", err)
+	}
+
+	fmt.Printf("load collection done, time elasped: %v\n", time.Since(start))
+	fmt.Printf(msgFmt, "start searching based on vector similarity")
+
+	vec2search := []entity.Vector{
+		entity.FloatVector(embeddingList[len(embeddingList)-2]),
+		entity.FloatVector(embeddingList[len(embeddingList)-1]),
+	}
+	begin := time.Now()
+	sp, _ := entity.NewIndexIvfFlatSearchParam(16)
+	sRet, err := c.Search(ctx, collectionName, nil, "", []string{ /*randomCol, typeCol*/ "*"}, vec2search,
+		embeddingCol, entity.L2, topK, sp)
+	end := time.Now()
+	if err != nil {
+		log.Fatalf("failed to search collection, err: %v", err)
+	}
+
+	fmt.Println("results:")
+	for _, res := range sRet {
+		printResult(&res)
+	}
+	fmt.Printf("\tsearch latency: %dms\n", end.Sub(begin)/time.Millisecond)
+
+	// hybrid search
+	fmt.Printf(msgFmt, "start hybrid searching with `random > 0.9`")
+	begin = time.Now()
+	sRet2, err := c.Search(ctx, collectionName, nil, "random > 0.9",
+		[]string{randomCol, typeCol}, vec2search, embeddingCol, entity.L2, topK, sp)
+	end = time.Now()
+	if err != nil {
+		log.Fatalf("failed to search collection, err: %v", err)
+	}
+	fmt.Println("results:")
+	for _, res := range sRet2 {
+		printResult(&res)
+	}
+	fmt.Printf("\tsearch latency: %dms\n", end.Sub(begin)/time.Millisecond)
+
+	// query
+	expr := "ID in [0, 1, 2]"
+	fmt.Printf(msgFmt, fmt.Sprintf("query with expr `%s`", expr))
+	sRet3, err := c.Query(ctx, collectionName, nil, expr, []string{randomCol, typeCol})
+	if err != nil {
+		log.Fatalf("failed to query result, err: %v", err)
+	}
+	printResultSet(sRet3)
+
+	// $meta["source"]
+	expr = "source in [1] and random > 0.1"
+	fmt.Printf(msgFmt, fmt.Sprintf("query with expr `%s`", expr))
+	sRet3, err = c.Query(ctx, collectionName, nil, expr, []string{randomCol, typeCol, sourceCol}, client.WithLimit(3))
+	if err != nil {
+		log.Fatalf("failed to query result, err: %v", err)
+	}
+	printResultSet(sRet3)
+
+	// drop collection
+	fmt.Printf(msgFmt, "drop collection `dynamic_example`")
+	if err := c.DropCollection(ctx, collectionName); err != nil {
+		log.Fatalf("failed to drop collection, err: %v", err)
+	}
+}
+
+func printResultSet(sRets []entity.Column) {
+	for _, field := range sRets {
+		fmt.Println(field.Name(), ":")
+		if dc, ok := field.(*entity.ColumnDynamic); ok {
+			for i := 0; i < field.Len(); i++ {
+				switch dc.Name() {
+				case typeCol, sourceCol:
+					v, err := dc.GetInt64(i)
+					if err != nil {
+						fmt.Println(err.Error())
+						continue
+					}
+					fmt.Print(v)
+				case randomCol:
+					v, err := dc.GetDouble(i)
+					if err != nil {
+						fmt.Println(err.Error())
+						continue
+					}
+					fmt.Print(v)
+				default:
+					continue
+				}
+				if i != field.Len()-1 {
+					fmt.Print(", ")
+				}
+			}
+			fmt.Println()
+			continue
+		}
+		for i := 0; i < field.Len(); i++ {
+			v, _ := field.Get(i)
+			fmt.Print(v)
+			if i != field.Len()-1 {
+				fmt.Print(", ")
+			}
+		}
+		fmt.Println()
+	}
+}
+
+func printResult(sRet *client.SearchResult) {
+	printResultSet(sRet.Fields)
+}


Test patch:
diff --git a/entity/rows_test.go b/entity/rows_test.go
index ea8e48c..0abe34b 100644
--- a/entity/rows_test.go
+++ b/entity/rows_test.go
@@ -12,6 +12,7 @@
 package entity
 
 import (
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -66,6 +67,12 @@ func TestParseSchema(t *testing.T) {
 		assert.Nil(t, sch)
 		assert.NotNil(t, err)
 
+		// MapRow
+		m := make(MapRow)
+		sch, err = ParseSchema(m)
+		assert.Nil(t, sch)
+		assert.NotNil(t, err)
+
 		// non struct
 		arrayRow := ArrayRow([16]float32{})
 		sch, err = ParseSchema(&arrayRow)
@@ -263,6 +270,50 @@ func (s *RowsSuite) TestDynamicSchema() {
 	})
 }
 
+func (s *RowsSuite) TestReflectValueCandi() {
+	cases := []struct {
+		tag       string
+		v         reflect.Value
+		expect    map[string]fieldCandi
+		expectErr bool
+	}{
+		{
+			tag: "MapRow",
+			v: reflect.ValueOf(MapRow(map[string]interface{}{
+				"A": "abd", "B": int64(8),
+			})),
+			expect: map[string]fieldCandi{
+				"A": {
+					name: "A",
+					v:    reflect.ValueOf("abd"),
+				},
+				"B": {
+					name: "B",
+					v:    reflect.ValueOf(int64(8)),
+				},
+			},
+			expectErr: false,
+		},
+	}
+
+	for _, c := range cases {
+		s.Run(c.tag, func() {
+			r, err := reflectValueCandi(c.v)
+			if c.expectErr {
+				s.Error(err)
+				return
+			}
+			s.NoError(err)
+			s.Equal(len(c.expect), len(r))
+			for k, v := range c.expect {
+				rv, has := r[k]
+				s.Require().True(has)
+				s.Equal(v.name, rv.name)
+			}
+		})
+	}
+}
+
 func TestRows(t *testing.T) {
 	suite.Run(t, new(RowsSuite))
 }