fix: add the db information in the dml message (#37969)

- issue: #37966

Signed-off-by: SimFG <bang.fu@zilliz.com>

internal/proxy/meta_cache.go

@@ -42,6 +42,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/util"
 	"github.com/milvus-io/milvus/pkg/util/commonpbutil"
 	"github.com/milvus-io/milvus/pkg/util/conc"
+	"github.com/milvus-io/milvus/pkg/util/expr"
 	"github.com/milvus-io/milvus/pkg/util/funcutil"
 	"github.com/milvus-io/milvus/pkg/util/merr"
 	"github.com/milvus-io/milvus/pkg/util/paramtable"
@@ -350,6 +351,7 @@ func InitMetaCache(ctx context.Context, rootCoord types.RootCoordClient, queryCo
 	if err != nil {
 		return err
 	}
+	expr.Register("cache", globalMetaCache)
 
 	// The privilege info is a little more. And to get this info, the query operation of involving multiple table queries is required.
 	resp, err := rootCoord.ListPolicy(ctx, &internalpb.ListPolicyRequest{})

internal/proxy/msg_pack.go

@@ -58,6 +58,7 @@ func genInsertMsgsByPartition(ctx context.Context,
 			),
 			CollectionID:   insertMsg.CollectionID,
 			PartitionID:    partitionID,
+			DbName:         insertMsg.DbName,
 			CollectionName: insertMsg.CollectionName,
 			PartitionName:  partitionName,
 			SegmentID:      segmentID,

internal/proxy/task_delete.go

@@ -56,6 +56,7 @@ type deleteTask struct {
 	primaryKeys      *schemapb.IDs
 	collectionID     UniqueID
 	partitionID      UniqueID
+	dbID             UniqueID
 	partitionKeyMode bool
 
 	// set by scheduler
@@ -148,14 +149,11 @@ func (dt *deleteTask) Execute(ctx context.Context) (err error) {
 
 	result, numRows, err := repackDeleteMsgByHash(
 		ctx,
-		dt.primaryKeys,
+		dt.primaryKeys, dt.vChannels,
-		dt.vChannels,
+		dt.idAllocator, dt.ts,
-		dt.idAllocator,
+		dt.collectionID, dt.req.GetCollectionName(),
-		dt.ts,
+		dt.partitionID, dt.req.GetPartitionName(),
-		dt.collectionID,
+		dt.req.GetDbName(),
-		dt.req.GetCollectionName(),
-		dt.partitionID,
-		dt.req.GetPartitionName(),
 	)
 	if err != nil {
 		return err
@@ -204,6 +202,7 @@ func repackDeleteMsgByHash(
 	collectionName string,
 	partitionID int64,
 	partitionName string,
+	dbName string,
 ) (map[uint32][]*msgstream.DeleteMsg, int64, error) {
 	maxSize := Params.PulsarCfg.MaxMessageSize.GetAsInt()
 	hashValues := typeutil.HashPK2Channels(primaryKeys, vChannels)
@@ -233,6 +232,7 @@ func repackDeleteMsgByHash(
 				PartitionID:    partitionID,
 				CollectionName: collectionName,
 				PartitionName:  partitionName,
+				DbName:         dbName,
 				PrimaryKeys:    &schemapb.IDs{},
 				ShardName:      vchannel,
 			},
@@ -413,6 +413,7 @@ func (dr *deleteRunner) produce(ctx context.Context, primaryKeys *schemapb.IDs)
 		partitionKeyMode: dr.partitionKeyMode,
 		vChannels:        dr.vChannels,
 		primaryKeys:      primaryKeys,
+		dbID:             dr.dbID,
 	}
 	var enqueuedTask task = dt
 	if streamingutil.IsStreamingServiceEnabled() {

internal/proxy/task_delete_streaming.go

@@ -31,15 +31,12 @@ func (dt *deleteTaskByStreamingService) Execute(ctx context.Context) (err error)
 
 	dt.tr = timerecord.NewTimeRecorder(fmt.Sprintf("proxy execute delete %d", dt.ID()))
 	result, numRows, err := repackDeleteMsgByHash(
-		ctx,
+		ctx, dt.primaryKeys,
-		dt.primaryKeys,
+		dt.vChannels, dt.idAllocator,
-		dt.vChannels,
+		dt.ts, dt.collectionID,
-		dt.idAllocator,
-		dt.ts,
-		dt.collectionID,
 		dt.req.GetCollectionName(),
-		dt.partitionID,
+		dt.partitionID, dt.req.GetPartitionName(),
-		dt.req.GetPartitionName(),
+		dt.req.GetDbName(),
 	)
 	if err != nil {
 		return err

internal/proxy/task_upsert_streaming.go

@@ -109,13 +109,11 @@ func (it *upsertTaskByStreamingService) packDeleteMessage(ctx context.Context) (
 	result, numRows, err := repackDeleteMsgByHash(
 		ctx,
 		it.upsertMsg.DeleteMsg.PrimaryKeys,
-		vChannels,
+		vChannels, it.idAllocator,
-		it.idAllocator,
 		it.BeginTs(),
-		it.upsertMsg.DeleteMsg.CollectionID,
+		it.upsertMsg.DeleteMsg.CollectionID, it.upsertMsg.DeleteMsg.CollectionName,
-		it.upsertMsg.DeleteMsg.CollectionName,
+		it.upsertMsg.DeleteMsg.PartitionID, it.upsertMsg.DeleteMsg.PartitionName,
-		it.upsertMsg.DeleteMsg.PartitionID,
+		it.req.GetDbName(),
-		it.upsertMsg.DeleteMsg.PartitionName,
 	)
 	if err != nil {
 		return nil, err