Filter indexable fields in flush scheduler

Signed-off-by: Nsunby <bingyi.sun@zilliz.com>

internal/master/flush_scheduler.go

@@ -68,6 +68,18 @@ func (scheduler *FlushScheduler) describe() error {
 						return err
 					}
 					for fieldID, data := range mapData {
+						// check field indexable
+						segMeta, err := scheduler.metaTable.GetSegmentByID(singleSegmentID)
+						if err != nil {
+							return err
+						}
+						indexable, err := scheduler.metaTable.IsIndexable(segMeta.CollectionID, fieldID)
+						if err != nil {
+							return err
+						}
+						if !indexable {
+							continue
+						}
 						info := &IndexBuildInfo{
 							segmentID:      singleSegmentID,
 							fieldID:        fieldID,

internal/master/meta_table.go

@@ -5,6 +5,8 @@ import (
 	"strconv"
 	"sync"
 
+	"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
+
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 
 	"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
@@ -678,3 +680,23 @@ func (mt *metaTable) UpdateFieldIndexParams(collName string, fieldName string, i
 
 	return fmt.Errorf("can not find field with id %s", fieldName)
 }
+
+func (mt *metaTable) IsIndexable(collID UniqueID, fieldID UniqueID) (bool, error) {
+	mt.ddLock.RLock()
+	defer mt.ddLock.RUnlock()
+
+	if _, ok := mt.collID2Meta[collID]; !ok {
+		return false, fmt.Errorf("can not find collection with id %d", collID)
+	}
+
+	for _, v := range mt.collID2Meta[collID].Schema.Fields {
+		// field is vector type and index params is not empty
+		if v.FieldID == fieldID && (v.DataType == schemapb.DataType_VECTOR_BINARY || v.DataType == schemapb.DataType_VECTOR_FLOAT) &&
+			len(v.IndexParams) != 0 {
+			return true, nil
+		}
+	}
+
+	// fieldID is not in schema(eg: timestamp) or not indexable
+	return false, nil
+}

internal/master/persistence_scheduler_test.go

@@ -5,6 +5,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
+
 	"github.com/zilliztech/milvus-distributed/internal/proto/etcdpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
 
@@ -43,7 +45,7 @@ func TestPersistenceScheduler(t *testing.T) {
 			Name: "testcoll",
 			Fields: []*schemapb.FieldSchema{
 				{FieldID: 1},
-				{FieldID: 100},
+				{FieldID: 100, DataType: schemapb.DataType_VECTOR_FLOAT, IndexParams: []*commonpb.KeyValuePair{{Key: "k", Value: "v"}}},
 			},
 		},
 	})

tests/python/requirements.txt

@@ -4,5 +4,5 @@ numpy==1.18.1
 pytest==5.3.4
 pytest-cov==2.8.1
 pytest-timeout==1.3.4
-pymilvus-distributed==0.0.6
+pymilvus-distributed==0.0.7
 sklearn==0.0

tests/python/utils.py

@@ -243,7 +243,7 @@ def gen_single_vector_fields():
 def gen_default_fields(auto_id=True):
     default_fields = {
         "fields": [
-            {"name": "int64", "type": DataType.INT64, "is_primary_key": not auto_id},
+            {"name": "int64", "type": DataType.INT64},
             {"name": "float", "type": DataType.FLOAT},
             {"name": default_float_vec_field_name, "type": DataType.FLOAT_VECTOR,
              "params": {"dim": default_dim},
@@ -258,7 +258,7 @@ def gen_default_fields(auto_id=True):
 def gen_binary_default_fields(auto_id=True):
     default_fields = {
         "fields": [
-            {"name": "int64", "type": DataType.INT64, "is_primary_key": not auto_id},
+            {"name": "int64", "type": DataType.INT64},
             {"name": "float", "type": DataType.FLOAT},
             {"name": default_binary_vec_field_name, "type": DataType.BINARY_VECTOR, "params": {"dim": default_dim}, "indexes": [{"metric_type": "JACCARD"}]}
             ],