updaet

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>

internal/storage/insert_data.go

@@ -116,7 +116,11 @@ func (i *InsertData) Append(row map[FieldID]interface{}) error {
 }
 
 func (i *InsertData) GetRow(idx int) map[FieldID]interface{} {
-	return nil
+	res := make(map[FieldID]interface{})
+	for field, data := range i.Data {
+		res[field] = data.GetRow(idx)
+	}
+	return res
 }
 
 // FieldData defines field data interface

internal/util/importutilv2/binlog/column_reader.go

@@ -19,6 +19,7 @@ package binlog
 import (
 	"context"
 	"fmt"
+
 	"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
 	"github.com/milvus-io/milvus/internal/storage"
 	"github.com/milvus-io/milvus/internal/util/importutilv2"

internal/util/importutilv2/binlog/filter.go

@@ -23,7 +23,7 @@ import (
 
 type Filter func(row map[int64]interface{}) bool
 
-func FilterWithDelete(r *Reader) (Filter, error) {
+func FilterWithDelete(r *reader) (Filter, error) {
 	pkField, err := typeutil.GetPrimaryFieldSchema(r.schema)
 	if err != nil {
 		return nil, err
@@ -39,7 +39,7 @@ func FilterWithDelete(r *Reader) (Filter, error) {
 	}, nil
 }
 
-func FilterWithTimerange(r *Reader) Filter {
+func FilterWithTimerange(r *reader) Filter {
 	return func(row map[int64]interface{}) bool {
 		ts := row[common.TimeStampField].(int64)
 		return uint64(ts) >= r.tsBegin && uint64(ts) <= r.tsEnd

internal/util/importutilv2/binlog/reader.go

@@ -19,14 +19,15 @@ package binlog
 import (
 	"context"
 	"encoding/json"
+
 	"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
 	"github.com/milvus-io/milvus/internal/proto/datapb"
 	"github.com/milvus-io/milvus/internal/storage"
 	"github.com/milvus-io/milvus/pkg/util/merr"
 	"github.com/milvus-io/milvus/pkg/util/typeutil"
 )
 
-type Reader struct {
+type reader struct {
 	cm     storage.ChunkManager
 	schema *schemapb.CollectionSchema
 
@@ -40,7 +41,7 @@ type Reader struct {
 	tsEnd   uint64
 }
 
-func (r *Reader) Init(paths []string) error {
+func (r *reader) Init(paths []string) error {
 	insertLogs, deltaLogs, err := r.ListBinlogs(paths)
 	if err != nil {
 		return err
@@ -61,7 +62,7 @@ func (r *Reader) Init(paths []string) error {
 	return nil
 }
 
-func (r *Reader) ReadDelta(deltaLogs *datapb.FieldBinlog) (*storage.DeleteData, error) {
+func (r *reader) ReadDelta(deltaLogs *datapb.FieldBinlog) (*storage.DeleteData, error) {
 	deleteData := storage.NewDeleteData(nil, nil)
 	for _, binlog := range deltaLogs.GetBinlogs() {
 		path := binlog.GetLogPath()
@@ -87,7 +88,7 @@ func (r *Reader) ReadDelta(deltaLogs *datapb.FieldBinlog) (*storage.DeleteData,
 	return deleteData, nil
 }
 
-func (r *Reader) ListBinlogs(paths []string) ([]*datapb.FieldBinlog, *datapb.FieldBinlog, error) {
+func (r *reader) ListBinlogs(paths []string) ([]*datapb.FieldBinlog, *datapb.FieldBinlog, error) {
 	if len(paths) < 1 {
 		return nil, nil, merr.WrapErrImportFailed("no insert binlogs to import")
 	}
@@ -106,7 +107,7 @@ func (r *Reader) ListBinlogs(paths []string) ([]*datapb.FieldBinlog, *datapb.Fie
 	return nil, nil, nil
 }
 
-func (r *Reader) Next(count int64) (*storage.InsertData, error) {
+func (r *reader) Next(count int64) (*storage.InsertData, error) {
 	insertData, err := storage.NewInsertData(r.schema)
 	if err != nil {
 		return nil, err
@@ -147,7 +148,6 @@ func (r *Reader) Next(count int64) (*storage.InsertData, error) {
 			return nil, err
 		}
 	}
-
 	r.readIdx++
 	return res, nil
 }

internal/util/importutilv2/reader.go

@@ -21,12 +21,14 @@ import (
 )
 
 type Reader interface {
-	//ReadStats()
-	//Close()
+	// ReadStats()
+	// Close()
+
 	Next(count int64) (*storage.InsertData, error)
 }
 
 type ColumnReader interface {
-	//Close()
+	// Close()
+
 	Next(count int64) (storage.FieldData, error)
 }