mirror of
https://gitee.com/milvus-io/milvus.git
synced 2024-11-30 02:48:45 +08:00
Add pre-checks in create index task
Signed-off-by: sunby <bingyi.sun@zilliz.com>
This commit is contained in:
parent
92b2df14ca
commit
3b68bf22f3
@ -34,9 +34,6 @@ IndexWrapper::IndexWrapper(const char* serialized_type_params, const char* seria
|
||||
auto mode = get_config_by_name<std::string>("index_mode");
|
||||
auto index_mode = mode.has_value() ? mode_map[mode.value()] : knowhere::IndexMode::MODE_CPU;
|
||||
|
||||
auto index_type = get_index_type();
|
||||
auto metric_type = get_metric_type();
|
||||
AssertInfo(!is_unsupported(index_type, metric_type), index_type + " doesn't support metric: " + metric_type);
|
||||
index_ = knowhere::VecIndexFactory::GetInstance().CreateVecIndex(get_index_type(), index_mode);
|
||||
Assert(index_ != nullptr);
|
||||
}
|
||||
@ -266,21 +263,6 @@ IndexWrapper::get_index_type() {
|
||||
return type.has_value() ? type.value() : knowhere::IndexEnum::INDEX_FAISS_IVFPQ;
|
||||
}
|
||||
|
||||
std::string
|
||||
IndexWrapper::get_metric_type() {
|
||||
auto type = get_config_by_name<std::string>(knowhere::Metric::TYPE);
|
||||
if (type.has_value()) {
|
||||
return type.value();
|
||||
} else {
|
||||
auto index_type = get_index_type();
|
||||
if (is_in_bin_list(index_type)) {
|
||||
return knowhere::Metric::JACCARD;
|
||||
} else {
|
||||
return knowhere::Metric::L2;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
std::unique_ptr<IndexWrapper::QueryResult>
|
||||
IndexWrapper::Query(const knowhere::DatasetPtr& dataset) {
|
||||
return std::move(QueryImpl(dataset, config_));
|
||||
|
@ -59,9 +59,6 @@ class IndexWrapper {
|
||||
std::string
|
||||
get_index_type();
|
||||
|
||||
std::string
|
||||
get_metric_type();
|
||||
|
||||
template <typename T>
|
||||
std::optional<T>
|
||||
get_config_by_name(std::string name);
|
||||
|
@ -14,7 +14,6 @@
|
||||
#include <vector>
|
||||
#include <string>
|
||||
#include <algorithm>
|
||||
#include <tuple>
|
||||
|
||||
#include "index/knowhere/knowhere/index/IndexType.h"
|
||||
|
||||
@ -58,14 +57,6 @@ Need_BuildAll_list() {
|
||||
return ret;
|
||||
}
|
||||
|
||||
std::vector<std::tuple<std::string, std::string>>
|
||||
unsupported_index_combinations() {
|
||||
static std::vector<std::tuple<std::string, std::string>> ret{
|
||||
std::make_tuple(std::string(knowhere::IndexEnum::INDEX_FAISS_BIN_IVFFLAT), std::string(knowhere::Metric::L2)),
|
||||
};
|
||||
return ret;
|
||||
}
|
||||
|
||||
template <typename T>
|
||||
bool
|
||||
is_in_list(const T& t, std::function<std::vector<T>()> list_func) {
|
||||
@ -93,11 +84,5 @@ is_in_need_id_list(const milvus::knowhere::IndexType& index_type) {
|
||||
return is_in_list<std::string>(index_type, Need_ID_List);
|
||||
}
|
||||
|
||||
bool
|
||||
is_unsupported(const milvus::knowhere::IndexType& index_type, const milvus::knowhere::MetricType& metric_type) {
|
||||
return is_in_list<std::tuple<std::string, std::string>>(std::make_tuple(index_type, metric_type),
|
||||
unsupported_index_combinations);
|
||||
}
|
||||
|
||||
} // namespace indexbuilder
|
||||
} // namespace milvus
|
||||
|
@ -14,7 +14,6 @@ package indexbuilder
|
||||
import "C"
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"unsafe"
|
||||
|
||||
@ -143,8 +142,6 @@ func (index *CIndex) Delete() error {
|
||||
}
|
||||
|
||||
func NewCIndex(typeParams, indexParams map[string]string) (Index, error) {
|
||||
fmt.Println("NNNNNNNNNNNNNNNNNNNNNNNNNNN typeParams: ", typeParams)
|
||||
fmt.Println("NNNNNNNNNNNNNNNNNNNNNNNNNNN indexParams: ", indexParams)
|
||||
protoTypeParams := &indexcgopb.TypeParams{
|
||||
Params: make([]*commonpb.KeyValuePair, 0),
|
||||
}
|
||||
@ -171,14 +168,10 @@ func NewCIndex(typeParams, indexParams map[string]string) (Index, error) {
|
||||
CIndex* res_index);
|
||||
*/
|
||||
var indexPtr C.CIndex
|
||||
fmt.Println("before create index ........................................")
|
||||
status := C.CreateIndex(typeParamsPointer, indexParamsPointer, &indexPtr)
|
||||
fmt.Println("after create index ........................................")
|
||||
errorCode := status.error_code
|
||||
fmt.Println("EEEEEEEEEEEEEEEEEEEEEEEEEE error code: ", errorCode)
|
||||
if errorCode != 0 {
|
||||
errorMsg := C.GoString(status.error_msg)
|
||||
fmt.Println("EEEEEEEEEEEEEEEEEEEEEEEEEE error msg: ", errorMsg)
|
||||
defer C.free(unsafe.Pointer(status.error_msg))
|
||||
return nil, errors.New(" failed, C runtime error detected, error code = " + strconv.Itoa(int(errorCode)) + ", error msg = " + errorMsg)
|
||||
}
|
||||
|
@ -2,7 +2,6 @@ package indexbuilder
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"strconv"
|
||||
"time"
|
||||
@ -172,12 +171,10 @@ func (it *IndexBuildTask) Execute() error {
|
||||
indexParams[key] = value
|
||||
}
|
||||
|
||||
fmt.Println("before NewCIndex ..........................")
|
||||
it.index, err = NewCIndex(typeParams, indexParams)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Println("after NewCIndex ..........................")
|
||||
|
||||
getKeyByPathNaive := func(path string) string {
|
||||
// splitElements := strings.Split(path, "/")
|
||||
@ -226,7 +223,6 @@ func (it *IndexBuildTask) Execute() error {
|
||||
|
||||
for _, value := range insertData.Data {
|
||||
// TODO: BinaryVectorFieldData
|
||||
fmt.Println("before build index ..................................")
|
||||
floatVectorFieldData, fOk := value.(*storage.FloatVectorFieldData)
|
||||
if fOk {
|
||||
err = it.index.BuildFloatVecIndexWithoutIds(floatVectorFieldData.Data)
|
||||
@ -242,15 +238,12 @@ func (it *IndexBuildTask) Execute() error {
|
||||
return err
|
||||
}
|
||||
}
|
||||
fmt.Println("after build index ..................................")
|
||||
|
||||
if !fOk && !bOk {
|
||||
return errors.New("we expect FloatVectorFieldData or BinaryVectorFieldData")
|
||||
}
|
||||
|
||||
fmt.Println("before serialize .............................................")
|
||||
indexBlobs, err := it.index.Serialize()
|
||||
fmt.Println("after serialize .............................................")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
@ -24,11 +24,6 @@ func (task *createIndexTask) Ts() (Timestamp, error) {
|
||||
}
|
||||
|
||||
func (task *createIndexTask) Execute() error {
|
||||
// modify schema
|
||||
if err := task.mt.UpdateFieldIndexParams(task.req.CollectionName, task.req.FieldName, task.req.ExtraParams); err != nil {
|
||||
return err
|
||||
}
|
||||
// check if closed segment has the same index build history
|
||||
collMeta, err := task.mt.GetCollectionByName(task.req.CollectionName)
|
||||
if err != nil {
|
||||
return err
|
||||
@ -44,6 +39,20 @@ func (task *createIndexTask) Execute() error {
|
||||
return fmt.Errorf("can not find field name %s", task.req.FieldName)
|
||||
}
|
||||
|
||||
// pre checks
|
||||
isIndexable, err := task.mt.IsIndexable(collMeta.ID, fieldID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !isIndexable {
|
||||
return fmt.Errorf("field %s is not vector", task.req.FieldName)
|
||||
}
|
||||
|
||||
// modify schema
|
||||
if err := task.mt.UpdateFieldIndexParams(task.req.CollectionName, task.req.FieldName, task.req.ExtraParams); err != nil {
|
||||
return err
|
||||
}
|
||||
// check if closed segment has the same index build history
|
||||
for _, segID := range collMeta.SegmentIDs {
|
||||
segMeta, err := task.mt.GetSegmentByID(segID)
|
||||
if err != nil {
|
||||
|
@ -488,7 +488,9 @@ func TestProxy_CreateIndex(t *testing.T) {
|
||||
go func(group *sync.WaitGroup) {
|
||||
defer group.Done()
|
||||
createCollection(t, collName)
|
||||
createIndex(t, collName, fieldName)
|
||||
if i%2 == 0 {
|
||||
createIndex(t, collName, fieldName)
|
||||
}
|
||||
dropCollection(t, collName)
|
||||
// dropIndex(t, collectionName, fieldName, indexName)
|
||||
}(&wg)
|
||||
@ -510,7 +512,9 @@ func TestProxy_DescribeIndex(t *testing.T) {
|
||||
go func(group *sync.WaitGroup) {
|
||||
defer group.Done()
|
||||
createCollection(t, collName)
|
||||
createIndex(t, collName, fieldName)
|
||||
if i%2 == 0 {
|
||||
createIndex(t, collName, fieldName)
|
||||
}
|
||||
req := &servicepb.DescribeIndexRequest{
|
||||
CollectionName: collName,
|
||||
FieldName: fieldName,
|
||||
@ -539,7 +543,9 @@ func TestProxy_DescribeIndexProgress(t *testing.T) {
|
||||
go func(group *sync.WaitGroup) {
|
||||
defer group.Done()
|
||||
createCollection(t, collName)
|
||||
createIndex(t, collName, fieldName)
|
||||
if i%2 == 0 {
|
||||
createIndex(t, collName, fieldName)
|
||||
}
|
||||
req := &servicepb.DescribeIndexProgressRequest{
|
||||
CollectionName: collName,
|
||||
FieldName: fieldName,
|
||||
|
Loading…
Reference in New Issue
Block a user