milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: drop partition can not be successful if load failed[2.5] (#38874) 

fix https://github.com/milvus-io/milvus/issues/38649
pr: #38793
when partition load failed, the partition drop will also fail due to the
wrong error message

Signed-off-by: xiaofanluan <xiaofan.luan@zilliz.com>
pull/38918/head
Xiaofan 2025-01-02 09:56:53 +08:00 committed by GitHub
parent 6fa096eb39
commit a2c4cd59ce
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 20 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/proxy/task.go
View File

@ -1462,7 +1462,7 @@ func (t *dropPartitionTask) PreExecute(ctx context.Context) error {
return err
}
if collLoaded {
loaded, err := isPartitionLoaded(ctx, t.queryCoord, collID, []int64{partID})
loaded, err := isPartitionLoaded(ctx, t.queryCoord, collID, partID)
if err != nil {
return err
}

7
internal/proxy/util.go
View File

@ -49,7 +49,6 @@ import (
"github.com/milvus-io/milvus/pkg/util/commonpbutil"
"github.com/milvus-io/milvus/pkg/util/contextutil"
"github.com/milvus-io/milvus/pkg/util/crypto"
"github.com/milvus-io/milvus/pkg/util/funcutil"
"github.com/milvus-io/milvus/pkg/util/merr"
"github.com/milvus-io/milvus/pkg/util/metric"
"github.com/milvus-io/milvus/pkg/util/paramtable"
@ -1476,11 +1475,11 @@ func isCollectionLoaded(ctx context.Context, qc types.QueryCoordClient, collID i
return false, nil
}
func isPartitionLoaded(ctx context.Context, qc types.QueryCoordClient, collID int64, partIDs []int64) (bool, error) {
func isPartitionLoaded(ctx context.Context, qc types.QueryCoordClient, collID int64, partID int64) (bool, error) {
// get all loading collections
resp, err := qc.ShowPartitions(ctx, &querypb.ShowPartitionsRequest{
CollectionID: collID,
PartitionIDs: partIDs,
PartitionIDs: []int64{partID},
})
if err := merr.CheckRPCCall(resp, err); err != nil {
// qc returns error if partition not loaded
@ -1490,7 +1489,7 @@ func isPartitionLoaded(ctx context.Context, qc types.QueryCoordClient, collID in
return false, err
}
return funcutil.SliceSetEqual(partIDs, resp.GetPartitionIDs()), nil
return true, nil
}
func checkFieldsDataBySchema(schema *schemapb.CollectionSchema, insertMsg *msgstream.InsertMsg, inInsert bool) error {

6
internal/proxy/util_test.go
View File

@ -1063,7 +1063,7 @@ func Test_isPartitionIsLoaded(t *testing.T) {
Status: merr.Success(),
PartitionIDs: []int64{partID},
}, nil)
loaded, err := isPartitionLoaded(ctx, qc, collID, []int64{partID})
loaded, err := isPartitionLoaded(ctx, qc, collID, partID)
assert.NoError(t, err)
assert.True(t, loaded)
})
@ -1088,7 +1088,7 @@ func Test_isPartitionIsLoaded(t *testing.T) {
Status: merr.Success(),
PartitionIDs: []int64{partID},
}, errors.New("error"))
loaded, err := isPartitionLoaded(ctx, qc, collID, []int64{partID})
loaded, err := isPartitionLoaded(ctx, qc, collID, partID)
assert.Error(t, err)
assert.False(t, loaded)
})
@ -1116,7 +1116,7 @@ func Test_isPartitionIsLoaded(t *testing.T) {
},
PartitionIDs: []int64{partID},
}, nil)
loaded, err := isPartitionLoaded(ctx, qc, collID, []int64{partID})
loaded, err := isPartitionLoaded(ctx, qc, collID, partID)
assert.Error(t, err)
assert.False(t, loaded)
})

7
internal/querycoordv2/services.go
View File

@ -159,15 +159,16 @@ func (s *Server) ShowPartitions(ctx context.Context, req *querypb.ShowPartitions
if percentage < 0 {
err := meta.GlobalFailedLoadCache.Get(req.GetCollectionID())
if err != nil {
status := merr.Status(err)
log.Warn("show partition failed", zap.Error(err))
partitionErr := merr.WrapErrPartitionNotLoaded(partitionID, err.Error())
status := merr.Status(partitionErr)
log.Warn("show partition failed", zap.Error(partitionErr))
return &querypb.ShowPartitionsResponse{
Status: status,
}, nil
}
err = merr.WrapErrPartitionNotLoaded(partitionID)
log.Warn("show partitions failed", zap.Error(err))
log.Warn("show partition failed", zap.Error(err))
return &querypb.ShowPartitionsResponse{
Status: merr.Status(err),
}, nil

7
internal/querycoordv2/services_test.go
View File

@ -24,6 +24,7 @@ import (
"github.com/cockroachdb/errors"
"github.com/samber/lo"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/suite"
@ -310,7 +311,8 @@ func (suite *ServiceSuite) TestShowPartitions() {
meta.GlobalFailedLoadCache.Put(collection, merr.WrapErrServiceMemoryLimitExceeded(100, 10))
resp, err = server.ShowPartitions(ctx, req)
suite.NoError(err)
suite.Equal(commonpb.ErrorCode_InsufficientMemoryToLoad, resp.GetStatus().GetErrorCode())
err := merr.CheckRPCCall(resp, err)
assert.True(suite.T(), errors.Is(err, merr.ErrPartitionNotLoaded))
meta.GlobalFailedLoadCache.Remove(collection)
err = suite.meta.CollectionManager.PutCollection(ctx, colBak)
suite.NoError(err)
@ -322,7 +324,8 @@ func (suite *ServiceSuite) TestShowPartitions() {
meta.GlobalFailedLoadCache.Put(collection, merr.WrapErrServiceMemoryLimitExceeded(100, 10))
resp, err = server.ShowPartitions(ctx, req)
suite.NoError(err)
suite.Equal(commonpb.ErrorCode_InsufficientMemoryToLoad, resp.GetStatus().GetErrorCode())
err := merr.CheckRPCCall(resp, err)
assert.True(suite.T(), errors.Is(err, merr.ErrPartitionNotLoaded))
meta.GlobalFailedLoadCache.Remove(collection)
err = suite.meta.CollectionManager.PutPartition(ctx, parBak)
suite.NoError(err)

8
pkg/util/merr/utils.go
View File

@ -40,14 +40,14 @@ func Code(err error) int32 {
}
cause := errors.Cause(err)
switch cause := cause.(type) {
switch specificErr := cause.(type) {
case milvusError:
return cause.code()
return specificErr.code()
default:
if errors.Is(cause, context.Canceled) {
if errors.Is(specificErr, context.Canceled) {
return CanceledCode
} else if errors.Is(cause, context.DeadlineExceeded) {
} else if errors.Is(specificErr, context.DeadlineExceeded) {
return TimeoutCode
} else {
return errUnexpected.code()