fix: marshal/unmarshal error

internal/net/grpc/stream.go

@@ -218,7 +218,7 @@ func BidirectionalStreamClient[S, R any](
 					return eg.Wait()
 				}
 
-				err = stream.SendMsg(data)
+				err = stream.SendMsg(*data)
 				if err != nil {
 					return err
 				}

pkg/tools/cli/loadtest/service/insert.go

@@ -57,8 +57,7 @@ func objectVectorProvider(dataset assets.Dataset) (func() *any, int) {
 					Vector: v.([]float32),
 				},
 			})
-			tmp := any(obj)
-			return &tmp
+			ret = &obj
 		}
 		return ret
 	}, size
@@ -77,20 +76,15 @@ func objectVectorsProvider(dataset assets.Dataset, n int) (func() *any, int) {
 			if d == nil {
 				break
 			}
-			insertRequest, ok := (*d).(*payload.Insert_Request)
-			if !ok {
-				return nil
-			}
-			r = append(r, insertRequest)
+			r = append(r, (*d).(*payload.Insert_Request))
 		}
 		if len(r) == 0 {
 			return nil
 		}
 		obj := any(&payload.Insert_MultiRequest{
 			Requests: r,
 		})
-		tmp := any(obj)
-		return &tmp
+		return &obj
 	}, size
 }
 

pkg/tools/cli/loadtest/service/loader.go

@@ -21,6 +21,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/vdaas/vald/apis/grpc/v1/payload"
 	"github.com/vdaas/vald/internal/errors"
 	"github.com/vdaas/vald/internal/log"
 	"github.com/vdaas/vald/internal/net/grpc"
@@ -201,7 +202,16 @@ func (l *loader) do(
 				if err != nil {
 					return nil, err
 				}
-				return nil, grpc.BidirectionalStreamClient(st.(grpc.ClientStream), l.sendDataProvider, f)
+
+				if l.operation == config.StreamInsert {
+					return nil, grpc.BidirectionalStreamClient(st.(grpc.ClientStream), l.sendDataProvider, func(i *payload.Empty, err error) {
+						f(nil, err)
+					})
+				} else {
+					return nil, grpc.BidirectionalStreamClient(st.(grpc.ClientStream), l.sendDataProvider, func(i *payload.Search_Response, err error) {
+						f(nil, err)
+					})
+				}
 			})
 			return err
 		}))