enhance: general optimizations (#49381)

* Optimize metadata and request handling hot paths
* Make replication pending-message queue capacity and max size
configurable instead of hard-coded.
* Batch QueryCoord collection metadata saves with MultiSave to avoid
oversized single writes while still reducing per-key etcd operations.
* Reduce avoidable allocations in DataCoord catalog tests, access log
list formatting, HTTP array joining, and HTTP JSON field extraction.
* Use WAL-specific message ID decoding during flusher recovery.
* Use set-based RootCoord collection-name filtering and structured
collection-not-found errors.

issue: #44452

Signed-off-by: Buqian Zheng <zhengbuqian@gmail.com>

configs/milvus.yaml

@@ -1397,6 +1397,9 @@ streaming:
     # When the wal is on-closing, the recovery module will try to persist the recovery info for wal to make next recovery operation more fast.
     # If that persist operation exceeds this timeout, the wal recovery module will close right now.
     gracefulCloseTimeout: 3s
+  replication:
+    pendingMessagesQueueLength: 128 # The capacity of pending message queue for each replication stream client.
+    pendingMessagesQueueMaxSize: 134217728 # The maximum size (in bytes) of pending message queue for each replication stream client. Default is 128MB.
 
 # Any configuration related to the knowhere vector search engine
 knowhere:

internal/cdc/replication/replicatestream/replicate_stream_client_impl.go

@@ -40,12 +40,6 @@ import (
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
-const (
-	// TODO: sheep, make these parameters configurable
-	pendingMessageQueueLength  = 128
-	pendingMessageQueueMaxSize = 128 * 1024 * 1024
-)
-
 var ErrReplicationRemoved = errors.New("replication removed")
 
 // replicateStreamClient is the implementation of ReplicateStreamClient.
@@ -68,8 +62,8 @@ func NewReplicateStreamClient(ctx context.Context, c cluster.MilvusClient, chann
 	ctx1 = contextutil.WithClusterID(ctx1, channel.Value.GetTargetCluster().GetClusterId())
 
 	options := MsgQueueOptions{
-		Capacity: pendingMessageQueueLength,
-		MaxSize:  pendingMessageQueueMaxSize,
+		Capacity: paramtable.Get().StreamingCfg.ReplicationPendingMessagesQueueLength.GetAsInt(),
+		MaxSize:  paramtable.Get().StreamingCfg.ReplicationPendingMessagesQueueMaxSize.GetAsInt(),
 	}
 	pendingMessages := NewMsgQueue(options)
 	rs := &replicateStreamClient{

internal/cdc/replication/replicatestream/replicate_stream_client_impl_test.go

@@ -40,6 +40,7 @@ import (
 	streamingpb "github.com/milvus-io/milvus/pkg/v3/proto/streamingpb"
 	message "github.com/milvus-io/milvus/pkg/v3/streaming/util/message"
 	"github.com/milvus-io/milvus/pkg/v3/streaming/walimpls/impls/walimplstest"
+	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
 func TestReplicateStreamClient_Replicate(t *testing.T) {
@@ -76,7 +77,7 @@ func TestReplicateStreamClient_Replicate(t *testing.T) {
 	defer replicateClient.Close()
 
 	// Test Replicate method
-	const msgCount = pendingMessageQueueLength * 10
+	msgCount := paramtable.Get().StreamingCfg.ReplicationPendingMessagesQueueLength.GetAsInt() * 10
 	go func() {
 		for i := 0; i < msgCount; i++ {
 			mockMsg := mock_message.NewMockImmutableMessage(t)

internal/distributed/proxy/httpserver/utils.go

@@ -17,7 +17,6 @@
 package httpserver
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"math"
@@ -110,18 +109,38 @@ func getPrimaryField(schema *schemapb.CollectionSchema) (*schemapb.FieldSchema,
 }
 
 func joinArray(data interface{}) string {
-	var buffer bytes.Buffer
-	arr := reflect.ValueOf(data)
-
-	for i := 0; i < arr.Len(); i++ {
-		if i > 0 {
-			buffer.WriteString(",")
-		}
-
-		fmt.Fprintf(&buffer, "%v", arr.Index(i))
+	if data == nil {
+		return ""
 	}
-
-	return buffer.String()
+	var builder strings.Builder
+	switch arr := data.(type) {
+	case []int64:
+		for i, v := range arr {
+			if i > 0 {
+				builder.WriteString(",")
+			}
+			builder.WriteString(strconv.FormatInt(v, 10))
+		}
+	case []string:
+		for i, v := range arr {
+			if i > 0 {
+				builder.WriteString(",")
+			}
+			builder.WriteString(v)
+		}
+	default:
+		rv := reflect.ValueOf(data)
+		if rv.Kind() != reflect.Slice && rv.Kind() != reflect.Array {
+			return ""
+		}
+		for i := 0; i < rv.Len(); i++ {
+			if i > 0 {
+				builder.WriteString(",")
+			}
+			fmt.Fprintf(&builder, "%v", rv.Index(i))
+		}
+	}
+	return builder.String()
 }
 
 func convertRange(field *schemapb.FieldSchema, result gjson.Result) (string, error) {
@@ -389,10 +408,10 @@ func checkAndSetData(body []byte, collSchema *schemapb.CollectionSchema, partial
 				}
 				fieldType := field.DataType
 				fieldName := field.Name
+				fieldValue := data.Get(fieldName)
 
 				if field.Nullable || field.DefaultValue != nil {
-					value := gjson.Get(data.Raw, fieldName)
-					if value.Type == gjson.Null {
+					if fieldValue.Type == gjson.Null {
 						validDataMap[fieldName] = append(validDataMap[fieldName], false)
 						continue
 					} else {
@@ -401,7 +420,6 @@ func checkAndSetData(body []byte, collSchema *schemapb.CollectionSchema, partial
 				}
 
 				// For partial update, check if field exists in the data
-				fieldValue := gjson.Get(data.Raw, fieldName)
 				if partialUpdate && !fieldValue.Exists() {
 					// Skip fields that are not provided in partial update
 					continue
@@ -434,7 +452,7 @@ func checkAndSetData(body []byte, collSchema *schemapb.CollectionSchema, partial
 					if dataString == "" {
 						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], "", "missing vector field: "+fieldName)
 					}
-					vectorStr := gjson.Get(data.Raw, fieldName).Raw
+					vectorStr := fieldValue.Raw
 					var vectorArray []byte
 					err := json.Unmarshal([]byte(vectorStr), &vectorArray)
 					if err != nil {
@@ -456,7 +474,7 @@ func checkAndSetData(body []byte, collSchema *schemapb.CollectionSchema, partial
 					if dataString == "" {
 						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], "", "missing vector field: "+fieldName)
 					}
-					vectorJSON := gjson.Get(data.Raw, fieldName)
+					vectorJSON := fieldValue
 					// Clients may send float32 vector because they are inconvenient of processing float16 or bfloat16.
 					// Float32 vector is an array in JSON format, like `[1.0, 2.0, 3.0]`, `[1, 2, 3]`, etc,
 					// while float16 or bfloat16 vector is a string in JSON format, like `"4z1jPgAAgL8="`, `"gD+AP4A/gD8="`, etc.
@@ -472,7 +490,7 @@ func checkAndSetData(body []byte, collSchema *schemapb.CollectionSchema, partial
 					} else if vectorJSON.Type == gjson.String {
 						// `data` is a float16 or bfloat16 vector
 						// same as `case schemapb.DataType_BinaryVector`
-						vectorStr := gjson.Get(data.Raw, fieldName).Raw
+						vectorStr := fieldValue.Raw
 						var vectorArray []byte
 						err := json.Unmarshal([]byte(vectorStr), &vectorArray)
 						if err != nil {

internal/metastore/kv/datacoord/kv_catalog_test.go

@@ -495,7 +495,7 @@ func Test_AlterSegments(t *testing.T) {
 		err := catalog.AlterSegments(context.TODO(), []*datapb.SegmentInfo{})
 		assert.NoError(t, err)
 
-		var binlogXL []*datapb.FieldBinlog
+		binlogXL := make([]*datapb.FieldBinlog, 0, 255)
 		for i := 0; i < 255; i++ {
 			binlogXL = append(binlogXL, &datapb.FieldBinlog{
 				FieldID: int64(i),

internal/metastore/kv/querycoord/kv_catalog.go

@@ -65,17 +65,25 @@ func (s Catalog) SaveCollection(ctx context.Context, collection *querypb.Collect
 }
 
 func (s Catalog) SavePartition(ctx context.Context, info ...*querypb.PartitionLoadInfo) error {
+	kvs := make(map[string]string)
 	for _, partition := range info {
 		k := EncodePartitionLoadInfoKey(partition.GetCollectionID(), partition.GetPartitionID())
 		v, err := proto.Marshal(partition)
 		if err != nil {
 			return err
 		}
-		err = s.cli.Save(ctx, k, string(v))
-		if err != nil {
-			return err
+		kvs[k] = string(v)
+		if len(kvs) >= MetaOpsBatchSize {
+			err := s.cli.MultiSave(ctx, kvs)
+			if err != nil {
+				return err
+			}
+			kvs = make(map[string]string)
 		}
 	}
+	if len(kvs) > 0 {
+		return s.cli.MultiSave(ctx, kvs)
+	}
 	return nil
 }
 

internal/proxy/accesslog/info/util.go

@@ -127,14 +127,21 @@ func getLengthFromTemplateValue(tv *schemapb.TemplateValue) int {
 }
 
 func listToString(strs []string) string {
-	result := "["
+	if len(strs) == 0 {
+		return "[]"
+	}
+	var b strings.Builder
+	b.WriteByte('[')
 	for i, str := range strs {
 		if i != 0 {
-			result += ", "
+			b.WriteString(", ")
 		}
-		result += "\"" + str + "\""
+		b.WriteByte('"')
+		b.WriteString(str)
+		b.WriteByte('"')
 	}
-	return result + "]"
+	b.WriteByte(']')
+	return b.String()
 }
 
 func kvsToString(kvs []*commonpb.KeyValuePair) string {

internal/proxy/accesslog/info/util_test.go

@@ -207,3 +207,17 @@ func TestGetCurUserFromContext(t *testing.T) {
 		assert.Error(t, err)
 	})
 }
+
+func TestListToString(t *testing.T) {
+	t.Run("empty list", func(t *testing.T) {
+		assert.Equal(t, "[]", listToString([]string{}))
+	})
+
+	t.Run("single element", func(t *testing.T) {
+		assert.Equal(t, "[\"foo\"]", listToString([]string{"foo"}))
+	})
+
+	t.Run("multiple elements", func(t *testing.T) {
+		assert.Equal(t, "[\"foo\", \"bar\"]", listToString([]string{"foo", "bar"}))
+	})
+}

internal/rootcoord/meta_table.go

@@ -797,7 +797,6 @@ func (mt *MetaTable) getCollectionByIDInternal(ctx context.Context, dbName strin
 	}
 
 	if coll == nil {
-		// use coll.Name to match error message of regression. TODO: remove this after error code is ready.
 		return nil, merr.WrapErrCollectionNotFound(collectionID)
 	}
 
@@ -806,8 +805,7 @@ func (mt *MetaTable) getCollectionByIDInternal(ctx context.Context, dbName strin
 	}
 
 	if !coll.Available() {
-		// use coll.Name to match error message of regression. TODO: remove this after error code is ready.
-		return nil, merr.WrapErrCollectionNotFound(dbName, coll.Name)
+		return nil, merr.WrapErrCollectionNotFound(collectionID)
 	}
 
 	return filterUnavailablePartition(coll), nil

internal/rootcoord/show_collection_task.go

@@ -19,8 +19,6 @@ package rootcoord
 import (
 	"context"
 
-	"github.com/samber/lo"
-
 	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/v3/milvuspb"
 	"github.com/milvus-io/milvus/pkg/v3/util/merr"
@@ -64,8 +62,12 @@ func (t *showCollectionTask) Execute(ctx context.Context) error {
 		t.Rsp.Status = merr.Status(err)
 		return err
 	}
+	var requestedCollections typeutil.Set[string]
+	if names := t.Req.GetCollectionNames(); len(names) > 0 {
+		requestedCollections = typeutil.NewSet(names...)
+	}
 	for _, coll := range colls {
-		if len(t.Req.GetCollectionNames()) > 0 && !lo.Contains(t.Req.GetCollectionNames(), coll.Name) {
+		if requestedCollections != nil && !requestedCollections.Contain(coll.Name) {
 			continue
 		}
 		if !isVisibleCollectionForCurUser(coll.Name, visibleCollections) {

internal/streamingnode/server/flusher/flusherimpl/util.go

@@ -43,8 +43,9 @@ func (impl *WALFlusherImpl) getRecoveryInfos(ctx context.Context, vchannel []str
 	}
 
 	var checkpoint message.MessageID
+	walName := impl.wal.Get().WALName()
 	for _, info := range recoveryInfos {
-		messageID := adaptor.MustGetMessageIDFromMQWrapperIDBytesWithWALName(impl.wal.Get().WALName(), info.GetInfo().GetSeekPosition().GetMsgID())
+		messageID := adaptor.MustGetMessageIDFromMQWrapperIDBytesWithWALName(walName, info.GetInfo().GetSeekPosition().GetMsgID())
 		if checkpoint == nil || messageID.LT(checkpoint) {
 			checkpoint = messageID
 		}

pkg/util/paramtable/component_param.go

@@ -7206,6 +7206,10 @@ type streamingConfig struct {
 
 	// Replication filtering configuration
 	ReplicationSkipMessageTypes ParamItem `refreshable:"false"`
+
+	// Replication pending message queue configuration
+	ReplicationPendingMessagesQueueLength  ParamItem `refreshable:"true"`
+	ReplicationPendingMessagesQueueMaxSize ParamItem `refreshable:"true"`
 }
 
 func (p *streamingConfig) init(base *BaseTable) {
@@ -7655,6 +7659,24 @@ so we set 1 second here as a threshold.`,
 	}
 	p.ReplicationSkipMessageTypes.Init(base.mgr)
 
+	p.ReplicationPendingMessagesQueueLength = ParamItem{
+		Key:          "streaming.replication.pendingMessagesQueueLength",
+		Version:      "3.0.0",
+		DefaultValue: "128",
+		Doc:          "The capacity of pending message queue for each replication stream client.",
+		Export:       true,
+	}
+	p.ReplicationPendingMessagesQueueLength.Init(base.mgr)
+
+	p.ReplicationPendingMessagesQueueMaxSize = ParamItem{
+		Key:          "streaming.replication.pendingMessagesQueueMaxSize",
+		Version:      "3.0.0",
+		DefaultValue: "134217728",
+		Doc:          "The maximum size (in bytes) of pending message queue for each replication stream client. Default is 128MB.",
+		Export:       true,
+	}
+	p.ReplicationPendingMessagesQueueMaxSize.Init(base.mgr)
+
 	p.WALRateLimitDefaultBurst = ParamItem{
 		Key:          "streaming.walRateLimit.defaultBurst",
 		Version:      "2.6.9",