Skip to content

Commit

Permalink
enhance: [2.4] autoindex for multi data type (milvus-io#33867)
Browse files Browse the repository at this point in the history
issue: milvus-io#22837 
pr: milvus-io#33868

- opensource autoindex support
- metric type check for different data types
- autoindex data type for search param

Signed-off-by: chasingegg <[email protected]>
  • Loading branch information
chasingegg authored Jun 14, 2024
1 parent 59144d8 commit 5fc1370
Show file tree
Hide file tree
Showing 14 changed files with 205 additions and 43 deletions.
44 changes: 32 additions & 12 deletions internal/proxy/task_index.go
Original file line number Diff line number Diff line change
Expand Up @@ -200,9 +200,7 @@ func (cit *createIndexTask) parseIndexParams() error {
indexParamsMap[common.MetricTypeKey] = metricType
}
} else { // behavior change after 2.2.9, adapt autoindex logic here.
autoIndexConfig := Params.AutoIndexConfig.IndexParams.GetAsJSONMap()

useAutoIndex := func() {
useAutoIndex := func(autoIndexConfig map[string]string) {
fields := make([]zap.Field, 0, len(autoIndexConfig))
for k, v := range autoIndexConfig {
indexParamsMap[k] = v
Expand All @@ -211,13 +209,13 @@ func (cit *createIndexTask) parseIndexParams() error {
log.Ctx(cit.ctx).Info("AutoIndex triggered", fields...)
}

handle := func(numberParams int) error {
handle := func(numberParams int, autoIndexConfig map[string]string) error {
// empty case.
if len(indexParamsMap) == numberParams {
// though we already know there must be metric type, how to make this safer to avoid crash?
metricType := autoIndexConfig[common.MetricTypeKey]
cit.newExtraParams = wrapUserIndexParams(metricType)
useAutoIndex()
useAutoIndex(autoIndexConfig)
return nil
}

Expand All @@ -234,20 +232,31 @@ func (cit *createIndexTask) parseIndexParams() error {

// only metric type is passed.
cit.newExtraParams = wrapUserIndexParams(metricType)
useAutoIndex()
useAutoIndex(autoIndexConfig)
// make the users' metric type first class citizen.
indexParamsMap[common.MetricTypeKey] = metricType
}

return nil
}

var config map[string]string
if typeutil.IsDenseFloatVectorType(cit.fieldSchema.DataType) {
// override float vector index params by autoindex
config = Params.AutoIndexConfig.IndexParams.GetAsJSONMap()
} else if typeutil.IsSparseFloatVectorType(cit.fieldSchema.DataType) {
// override sparse float vector index params by autoindex
config = Params.AutoIndexConfig.SparseIndexParams.GetAsJSONMap()
} else if typeutil.IsBinaryVectorType(cit.fieldSchema.DataType) {
// override binary vector index params by autoindex
config = Params.AutoIndexConfig.BinaryIndexParams.GetAsJSONMap()
}
if !exist {
if err := handle(0); err != nil {
if err := handle(0, config); err != nil {
return err
}
} else if specifyIndexType == AutoIndexName {
if err := handle(1); err != nil {
if err := handle(1, config); err != nil {
return err
}
}
Expand All @@ -263,10 +272,21 @@ func (cit *createIndexTask) parseIndexParams() error {
return err
}
}
if indexType == indexparamcheck.IndexSparseInverted || indexType == indexparamcheck.IndexSparseWand {
metricType, metricTypeExist := indexParamsMap[common.MetricTypeKey]
if !metricTypeExist || metricType != metric.IP {
return fmt.Errorf("only IP is the supported metric type for sparse index")
metricType, metricTypeExist := indexParamsMap[common.MetricTypeKey]
if !metricTypeExist {
return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "metric type not set for vector index")
}
if typeutil.IsDenseFloatVectorType(cit.fieldSchema.DataType) {
if !funcutil.SliceContain(indexparamcheck.FloatVectorMetrics, metricType) {
return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "float vector index does not support metric type: "+metricType)
}
} else if typeutil.IsSparseFloatVectorType(cit.fieldSchema.DataType) {
if metricType != metric.IP {
return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "only IP is the supported metric type for sparse index")
}
} else if typeutil.IsBinaryVectorType(cit.fieldSchema.DataType) {
if !funcutil.SliceContain(indexparamcheck.BinaryVectorMetrics, metricType) {
return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "binary vector index does not support metric type: "+metricType)
}
}
}
Expand Down
53 changes: 50 additions & 3 deletions internal/proxy/task_index_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -970,8 +970,8 @@ func Test_parseIndexParams_AutoIndex_WithType(t *testing.T) {
Params.AutoIndexConfig.Enable.Init(mgr)

mgr.SetConfig("autoIndex.params.build", `{"M": 30,"efConstruction": 360,"index_type": "HNSW"}`)
mgr.SetConfig("autoIndex.params.sparsebuild", `{"drop_ratio_build": 0.2, "index_type": "SPARSE_INVERTED_INDEX"}`)
mgr.SetConfig("autoIndex.params.binarybuild", `{"nlist": 1024, "index_type": "BIN_IVF_FLAT"}`)
mgr.SetConfig("autoIndex.params.sparse.build", `{"drop_ratio_build": 0.2, "index_type": "SPARSE_INVERTED_INDEX"}`)
mgr.SetConfig("autoIndex.params.binary.build", `{"nlist": 1024, "index_type": "BIN_IVF_FLAT"}`)
Params.AutoIndexConfig.IndexParams.Init(mgr)
Params.AutoIndexConfig.SparseIndexParams.Init(mgr)
Params.AutoIndexConfig.BinaryIndexParams.Init(mgr)
Expand Down Expand Up @@ -1057,17 +1057,64 @@ func Test_parseIndexParams_AutoIndex(t *testing.T) {
mgr := config.NewManager()
mgr.SetConfig("autoIndex.enable", "false")
mgr.SetConfig("autoIndex.params.build", `{"M": 30,"efConstruction": 360,"index_type": "HNSW", "metric_type": "IP"}`)
mgr.SetConfig("autoIndex.params.binary.build", `{"nlist": 1024, "index_type": "BIN_IVF_FLAT", "metric_type": "JACCARD"}`)
mgr.SetConfig("autoIndex.params.sparse.build", `{"index_type": "SPARSE_INVERTED_INDEX", "metric_type": "IP"}`)
Params.AutoIndexConfig.Enable.Init(mgr)
Params.AutoIndexConfig.IndexParams.Init(mgr)
Params.AutoIndexConfig.BinaryIndexParams.Init(mgr)
Params.AutoIndexConfig.SparseIndexParams.Init(mgr)
autoIndexConfig := Params.AutoIndexConfig.IndexParams.GetAsJSONMap()
autoIndexConfigBinary := Params.AutoIndexConfig.BinaryIndexParams.GetAsJSONMap()
autoIndexConfigSparse := Params.AutoIndexConfig.SparseIndexParams.GetAsJSONMap()
fieldSchema := &schemapb.FieldSchema{
DataType: schemapb.DataType_FloatVector,
TypeParams: []*commonpb.KeyValuePair{
{Key: common.DimKey, Value: "8"},
},
}

t.Run("case 1, empty parameters", func(t *testing.T) {
fieldSchemaBinary := &schemapb.FieldSchema{
DataType: schemapb.DataType_BinaryVector,
TypeParams: []*commonpb.KeyValuePair{
{Key: common.DimKey, Value: "8"},
},
}

fieldSchemaSparse := &schemapb.FieldSchema{
DataType: schemapb.DataType_SparseFloatVector,
}

t.Run("case 1, empty parameters binary", func(t *testing.T) {
task := &createIndexTask{
fieldSchema: fieldSchemaBinary,
req: &milvuspb.CreateIndexRequest{
ExtraParams: make([]*commonpb.KeyValuePair, 0),
},
}
err := task.parseIndexParams()
assert.NoError(t, err)
assert.ElementsMatch(t, []*commonpb.KeyValuePair{
{Key: common.IndexTypeKey, Value: AutoIndexName},
{Key: common.MetricTypeKey, Value: autoIndexConfigBinary[common.MetricTypeKey]},
}, task.newExtraParams)
})

t.Run("case 1, empty parameters sparse", func(t *testing.T) {
task := &createIndexTask{
fieldSchema: fieldSchemaSparse,
req: &milvuspb.CreateIndexRequest{
ExtraParams: make([]*commonpb.KeyValuePair, 0),
},
}
err := task.parseIndexParams()
assert.NoError(t, err)
assert.ElementsMatch(t, []*commonpb.KeyValuePair{
{Key: common.IndexTypeKey, Value: AutoIndexName},
{Key: common.MetricTypeKey, Value: autoIndexConfigSparse[common.MetricTypeKey]},
}, task.newExtraParams)
})

t.Run("case 1, empty parameters float vector", func(t *testing.T) {
task := &createIndexTask{
fieldSchema: fieldSchema,
req: &milvuspb.CreateIndexRequest{
Expand Down
1 change: 1 addition & 0 deletions internal/querynodev2/optimizers/query_hook.go
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ func OptimizeSearchParams(ctx context.Context, req *querypb.SearchRequest, query
common.SearchParamKey: queryInfo.GetSearchParams(),
common.SegmentNumKey: estSegmentNum,
common.WithFilterKey: withFilter,
common.DataTypeKey: plan.GetVectorAnns().GetVectorType(),
common.WithOptimizeKey: paramtable.Get().AutoIndexConfig.EnableOptimize.GetAsBool(),
common.CollectionKey: req.GetReq().GetCollectionID(),
}
Expand Down
1 change: 1 addition & 0 deletions pkg/common/common.go
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@ const (
SearchParamKey = "search_param"
SegmentNumKey = "segment_num"
WithFilterKey = "with_filter"
DataTypeKey = "data_type"
WithOptimizeKey = "with_optimize"
CollectionKey = "collection"

Expand Down
2 changes: 1 addition & 1 deletion pkg/util/indexparamcheck/base_checker.go
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ func (c baseChecker) CheckValidDataType(dType schemapb.DataType) error {
return nil
}

func (c baseChecker) SetDefaultMetricTypeIfNotExist(m map[string]string) {}
func (c baseChecker) SetDefaultMetricTypeIfNotExist(m map[string]string, dType schemapb.DataType) {}

func (c baseChecker) StaticCheck(params map[string]string) error {
return errors.New("unsupported index type")
Expand Down
2 changes: 1 addition & 1 deletion pkg/util/indexparamcheck/binary_vector_base_checker.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ func (c binaryVectorBaseChecker) CheckValidDataType(dType schemapb.DataType) err
return nil
}

func (c binaryVectorBaseChecker) SetDefaultMetricTypeIfNotExist(params map[string]string) {
func (c binaryVectorBaseChecker) SetDefaultMetricTypeIfNotExist(params map[string]string, dType schemapb.DataType) {
setDefaultIfNotExist(params, common.MetricTypeKey, BinaryVectorDefaultMetricType)
}

Expand Down
6 changes: 4 additions & 2 deletions pkg/util/indexparamcheck/constraints.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,10 @@ const (
SparseDropRatioBuild = "drop_ratio_build"
)

// METRICS is a set of all metrics types supported for float vector.
var METRICS = []string{metric.L2, metric.IP, metric.COSINE} // const
var (
FloatVectorMetrics = []string{metric.L2, metric.IP, metric.COSINE} // const
BinaryVectorMetrics = []string{metric.HAMMING, metric.JACCARD, metric.SUBSTRUCTURE, metric.SUPERSTRUCTURE} // const
)

// BinIDMapMetrics is a set of all metric types supported for binary vector.
var (
Expand Down
6 changes: 3 additions & 3 deletions pkg/util/indexparamcheck/float_vector_base_checker.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@ type floatVectorBaseChecker struct {
}

func (c floatVectorBaseChecker) staticCheck(params map[string]string) error {
if !CheckStrByValues(params, Metric, METRICS) {
return fmt.Errorf("metric type %s not found or not supported, supported: %v", params[Metric], METRICS)
if !CheckStrByValues(params, Metric, FloatVectorMetrics) {
return fmt.Errorf("metric type %s not found or not supported, supported: %v", params[Metric], FloatVectorMetrics)
}

return nil
Expand All @@ -35,7 +35,7 @@ func (c floatVectorBaseChecker) CheckValidDataType(dType schemapb.DataType) erro
return nil
}

func (c floatVectorBaseChecker) SetDefaultMetricTypeIfNotExist(params map[string]string) {
func (c floatVectorBaseChecker) SetDefaultMetricTypeIfNotExist(params map[string]string, dType schemapb.DataType) {
setDefaultIfNotExist(params, common.MetricTypeKey, FloatVectorDefaultMetricType)
}

Expand Down
13 changes: 12 additions & 1 deletion pkg/util/indexparamcheck/hnsw_checker.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,12 @@ import (
"fmt"

"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
"github.com/milvus-io/milvus/pkg/common"
"github.com/milvus-io/milvus/pkg/util/typeutil"
)

type hnswChecker struct {
floatVectorBaseChecker
baseChecker
}

func (c hnswChecker) StaticCheck(params map[string]string) error {
Expand Down Expand Up @@ -38,6 +39,16 @@ func (c hnswChecker) CheckValidDataType(dType schemapb.DataType) error {
return nil
}

func (c hnswChecker) SetDefaultMetricTypeIfNotExist(params map[string]string, dType schemapb.DataType) {
if typeutil.IsDenseFloatVectorType(dType) {
setDefaultIfNotExist(params, common.MetricTypeKey, FloatVectorDefaultMetricType)
} else if typeutil.IsSparseFloatVectorType(dType) {
setDefaultIfNotExist(params, common.MetricTypeKey, SparseFloatVectorDefaultMetricType)
} else if typeutil.IsBinaryVectorType(dType) {
setDefaultIfNotExist(params, common.MetricTypeKey, BinaryVectorDefaultMetricType)
}
}

func newHnswChecker() IndexChecker {
return &hnswChecker{}
}
39 changes: 39 additions & 0 deletions pkg/util/indexparamcheck/hnsw_checker_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -172,3 +172,42 @@ func Test_hnswChecker_CheckValidDataType(t *testing.T) {
}
}
}

func Test_hnswChecker_SetDefaultMetricType(t *testing.T) {
cases := []struct {
dType schemapb.DataType
metricType string
}{
{
dType: schemapb.DataType_FloatVector,
metricType: metric.IP,
},
{
dType: schemapb.DataType_Float16Vector,
metricType: metric.IP,
},
{
dType: schemapb.DataType_BFloat16Vector,
metricType: metric.IP,
},
{
dType: schemapb.DataType_SparseFloatVector,
metricType: metric.IP,
},
{
dType: schemapb.DataType_BinaryVector,
metricType: metric.JACCARD,
},
}

c := newHnswChecker()
for _, test := range cases {
p := map[string]string{
DIM: strconv.Itoa(128),
HNSWM: strconv.Itoa(16),
EFConstruction: strconv.Itoa(200),
}
c.SetDefaultMetricTypeIfNotExist(p, test.dType)
assert.Equal(t, p[Metric], test.metricType)
}
}
2 changes: 1 addition & 1 deletion pkg/util/indexparamcheck/index_checker.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,6 @@ import (
type IndexChecker interface {
CheckTrain(map[string]string) error
CheckValidDataType(dType schemapb.DataType) error
SetDefaultMetricTypeIfNotExist(map[string]string)
SetDefaultMetricTypeIfNotExist(map[string]string, schemapb.DataType)
StaticCheck(map[string]string) error
}
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ func (c sparseFloatVectorBaseChecker) CheckValidDataType(dType schemapb.DataType
return nil
}

func (c sparseFloatVectorBaseChecker) SetDefaultMetricTypeIfNotExist(params map[string]string) {
func (c sparseFloatVectorBaseChecker) SetDefaultMetricTypeIfNotExist(params map[string]string, dType schemapb.DataType) {
setDefaultIfNotExist(params, common.MetricTypeKey, SparseFloatVectorDefaultMetricType)
}

Expand Down
24 changes: 15 additions & 9 deletions pkg/util/paramtable/autoindex_param.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package paramtable
import (
"fmt"

"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
"github.com/milvus-io/milvus/pkg/common"
"github.com/milvus-io/milvus/pkg/config"
"github.com/milvus-io/milvus/pkg/util/funcutil"
Expand Down Expand Up @@ -193,31 +194,36 @@ func (p *autoIndexConfig) init(base *BaseTable) {
}

func (p *autoIndexConfig) panicIfNotValidAndSetDefaultMetricType(mgr *config.Manager) {
m := p.IndexParams.GetAsJSONMap()
p.panicIfNotValidAndSetDefaultMetricTypeHelper(p.IndexParams.Key, p.IndexParams.GetAsJSONMap(), schemapb.DataType_FloatVector, mgr)
p.panicIfNotValidAndSetDefaultMetricTypeHelper(p.BinaryIndexParams.Key, p.BinaryIndexParams.GetAsJSONMap(), schemapb.DataType_BinaryVector, mgr)
p.panicIfNotValidAndSetDefaultMetricTypeHelper(p.SparseIndexParams.Key, p.SparseIndexParams.GetAsJSONMap(), schemapb.DataType_SparseFloatVector, mgr)
}

func (p *autoIndexConfig) panicIfNotValidAndSetDefaultMetricTypeHelper(key string, m map[string]string, dtype schemapb.DataType, mgr *config.Manager) {
if m == nil {
panic("autoIndex.build not invalid, should be json format")
panic(fmt.Sprintf("%s invalid, should be json format", key))
}

indexType, ok := m[common.IndexTypeKey]
if !ok {
panic("autoIndex.build not invalid, index type not found")
panic(fmt.Sprintf("%s invalid, index type not found", key))
}

checker, err := indexparamcheck.GetIndexCheckerMgrInstance().GetChecker(indexType)
if err != nil {
panic(fmt.Sprintf("autoIndex.build not invalid, unsupported index type: %s", indexType))
panic(fmt.Sprintf("%s invalid, unsupported index type: %s", key, indexType))
}

checker.SetDefaultMetricTypeIfNotExist(m)
checker.SetDefaultMetricTypeIfNotExist(m, dtype)

if err := checker.StaticCheck(m); err != nil {
panic(fmt.Sprintf("autoIndex.build not invalid, parameters not invalid, error: %s", err.Error()))
panic(fmt.Sprintf("%s invalid, parameters invalid, error: %s", key, err.Error()))
}

p.reset(m, mgr)
p.reset(key, m, mgr)
}

func (p *autoIndexConfig) reset(m map[string]string, mgr *config.Manager) {
func (p *autoIndexConfig) reset(key string, m map[string]string, mgr *config.Manager) {
j := funcutil.MapToJSON(m)
mgr.SetConfig("autoIndex.params.build", string(j))
mgr.SetConfig(key, string(j))
}
Loading

0 comments on commit 5fc1370

Please sign in to comment.