Upsert raise error (#22533)

Signed-off-by: lixinguo <xinguo.li@zilliz.com>
Co-authored-by: lixinguo <xinguo.li@zilliz.com>

internal/proxy/impl.go

@@ -2341,6 +2341,11 @@ func (node *Proxy) Upsert(ctx context.Context, request *milvuspb.UpsertRequest)
 			zap.Error(err))
 		metrics.ProxyFunctionCall.WithLabelValues(strconv.FormatInt(paramtable.GetNodeID(), 10), method,
 			metrics.FailLabel).Inc()
+		// Not every error case changes the status internally
+		// change status there to handle it
+		if it.result.Status.ErrorCode == commonpb.ErrorCode_Success {
+			it.result.Status.ErrorCode = commonpb.ErrorCode_UnexpectedError
+		}
 		return constructFailedResponse(err, it.result.Status.ErrorCode), nil
 	}
 

internal/proxy/proxy_test.go

@@ -3333,6 +3333,21 @@ func TestProxy(t *testing.T) {
 		}
 	}
 
+	constructPartitionReqUpsertRequestInvalid := func() *milvuspb.UpsertRequest {
+		pkFieldData := newScalarFieldData(schema.Fields[0], int64Field, rowNum)
+		fVecColumn := newFloatVectorFieldData(floatVecField, rowNum, dim)
+		hashKeys := generateHashKeys(rowNum)
+		return &milvuspb.UpsertRequest{
+			Base:           nil,
+			DbName:         dbName,
+			CollectionName: collectionName,
+			PartitionName:  "%$@",
+			FieldsData:     []*schemapb.FieldData{pkFieldData, fVecColumn},
+			HashKeys:       hashKeys,
+			NumRows:        uint32(rowNum),
+		}
+	}
+
 	constructCollectionUpsertRequestValid := func() *milvuspb.UpsertRequest {
 		pkFieldData := newScalarFieldData(schema.Fields[0], int64Field, rowNum)
 		fVecColumn := newFloatVectorFieldData(floatVecField, rowNum, dim)
@@ -3409,6 +3424,19 @@ func TestProxy(t *testing.T) {
 		assert.Equal(t, int64(rowNum), resp.UpsertCnt)
 	})
 
+	wg.Add(1)
+	t.Run("upsert when occurs unexpected error like illegal partition name", func(t *testing.T) {
+		defer wg.Done()
+		req := constructPartitionReqUpsertRequestInvalid()
+
+		resp, err := proxy.Upsert(ctx, req)
+		assert.NoError(t, err)
+		assert.Equal(t, commonpb.ErrorCode_UnexpectedError, resp.Status.ErrorCode)
+		assert.Equal(t, 0, len(resp.SuccIndex))
+		assert.Equal(t, rowNum, len(resp.ErrIndex))
+		assert.Equal(t, int64(0), resp.UpsertCnt)
+	})
+
 	wg.Add(1)
 	t.Run("upsert when autoID == false", func(t *testing.T) {
 		defer wg.Done()