convert vectors from default to vfloat when rebuilding the index

Author: shivaji-kharse

posting/index.go

@@ -1373,7 +1373,8 @@ func rebuildVectorIndex(ctx context.Context, factorySpecs []*tok.FactoryCreateSp
 	}
 
 	dimension := indexer.Dimension()
-	if dimension == 0 {
+	// If dimension is -1, it means that the dimension is not set through options in case of partitioned hnsw.
+	if dimension == -1 {
 		numVectorsToCheck := 100
 		lenFreq := make(map[int]int, numVectorsToCheck)
 		maxFreq := 0
@@ -1410,6 +1411,48 @@ func rebuildVectorIndex(ctx context.Context, factorySpecs []*tok.FactoryCreateSp
 
 	fmt.Println("Selecting vector dimension to be:", dimension)
 
+	norm := rebuilder{attr: rb.Attr, prefix: pk.DataPrefix(), startTs: rb.StartTs}
+	norm.fn = func(uid uint64, pl *List, txn *Txn) ([]*pb.DirectedEdge, error) {
+		val, err := pl.Value(rb.StartTs)
+		if err != nil {
+			return nil, err
+		}
+		if val.Tid == types.VFloatID {
+			return nil, nil
+		}
+
+		// Convert to VFloatID and persist as binary bytes.
+		sv, err := types.Convert(val, types.VFloatID)
+		if err != nil {
+			return nil, err
+		}
+		b := types.ValueForType(types.BinaryID)
+		if err = types.Marshal(sv, &b); err != nil {
+			return nil, err
+		}
+
+		edge := &pb.DirectedEdge{
+			Attr:      rb.Attr,
+			Entity:    uid,
+			Value:     b.Value.([]byte),
+			ValueType: types.VFloatID.Enum(),
+		}
+		inKey := x.DataKey(edge.Attr, uid)
+		p, err := txn.Get(inKey)
+		if err != nil {
+			return []*pb.DirectedEdge{}, err
+		}
+
+		if err := p.addMutation(ctx, txn, edge); err != nil {
+			return []*pb.DirectedEdge{}, err
+		}
+		return nil, nil
+	}
+
+	if err := norm.RunWithoutTemp(ctx); err != nil {
+		return err
+	}
+
 	count := 0
 
 	if indexer.NumSeedVectors() > 0 {
@@ -1426,6 +1469,22 @@ func rebuildVectorIndex(ctx context.Context, factorySpecs []*tok.FactoryCreateSp
 				if err != nil {
 					return err
 				}
+
+				if val.Tid != types.VFloatID {
+					// Here, we convert the defaultID type vector into vfloat.
+					sv, err := types.Convert(val, types.VFloatID)
+					if err != nil {
+						return err
+					}
+					b := types.ValueForType(types.BinaryID)
+					if err = types.Marshal(sv, &b); err != nil {
+						return err
+					}
+
+					val.Value = b.Value
+					val.Tid = types.VFloatID
+				}
+
 				inVec := types.BytesAsFloatArray(val.Value.([]byte))
 				if len(inVec) != dimension {
 					return fmt.Errorf("vector dimension mismatch expected dimension %d but got %d", dimension, len(inVec))
@@ -1464,7 +1523,6 @@ func rebuildVectorIndex(ctx context.Context, factorySpecs []*tok.FactoryCreateSp
 
 		builder := rebuilder{attr: rb.Attr, prefix: pk.DataPrefix(), startTs: rb.StartTs}
 		builder.fn = func(uid uint64, pl *List, txn *Txn) ([]*pb.DirectedEdge, error) {
-			edges := []*pb.DirectedEdge{}
 			val, err := pl.Value(rb.StartTs)
 			if err != nil {
 				return []*pb.DirectedEdge{}, err
@@ -1475,7 +1533,7 @@ func rebuildVectorIndex(ctx context.Context, factorySpecs []*tok.FactoryCreateSp
 				return []*pb.DirectedEdge{}, nil
 			}
 			indexer.BuildInsert(ctx, uid, inVec)
-			return edges, nil
+			return []*pb.DirectedEdge{}, nil
 		}
 
 		err := builder.RunWithoutTemp(ctx)
@@ -1519,21 +1577,22 @@ func rebuildVectorIndex(ctx context.Context, factorySpecs []*tok.FactoryCreateSp
 
 		builder := rebuilder{attr: rb.Attr, prefix: pk.DataPrefix(), startTs: rb.StartTs}
 		builder.fn = func(uid uint64, pl *List, txn *Txn) ([]*pb.DirectedEdge, error) {
-			edges := []*pb.DirectedEdge{}
 			val, err := pl.Value(rb.StartTs)
 			if err != nil {
 				return []*pb.DirectedEdge{}, err
 			}
 
 			inVec := types.BytesAsFloatArray(val.Value.([]byte))
-			if len(inVec) != dimension {
+			if len(inVec) != dimension && centroids != nil {
 				if pass_idx == 0 {
 					glog.Warningf("Skipping vector with invalid dimension uid: %d, dimension: %d", uid, len(inVec))
 				}
 				return []*pb.DirectedEdge{}, nil
 			}
+
 			indexer.BuildInsert(ctx, uid, inVec)
-			return edges, nil
+
+			return []*pb.DirectedEdge{}, nil
 		}
 
 		err := builder.RunWithoutTemp(ctx)

systest/vector/vector_test.go

@@ -31,7 +31,7 @@ const (
 
 var schemas = map[string]string{
 	"hnsw":            `project_description_v: float32vector @index(hnsw(exponent: "5", metric: "euclidean")) .`,
-	"partitionedhnsw": `project_description_v: float32vector @index(partionedhnsw(numClusters: "100", partitionStratOpt: "kmeans", vectorDimension: "10", metric: "euclidean")) .`,
+	"partitionedhnsw": `project_description_v: float32vector @index(partionedhnsw(numClusters: "1000", partitionStratOpt: "kmeans", vectorDimension: "100", metric: "euclidean")) .`,
 }
 
 func testVectorQuery(t *testing.T, gc *dgraphapi.GrpcClient, vectors [][]float32, rdfs, pred string, topk int) {
@@ -431,6 +431,48 @@ func (vsuite *VectorTestSuite) TestVectorIndexWithoutSchema() {
 	require.JSONEq(t, fmt.Sprintf(`{"vector":[{"count":%v}]}`, numVectors), string(result.GetJson()))
 }
 
+func (vsuite *VectorTestSuite) TestIndexRebuildingWithoutSchema() {
+	t := vsuite.T()
+	conf := dgraphtest.NewClusterConfig().WithNumAlphas(1).WithNumZeros(1).WithReplicas(1).WithACL(time.Hour)
+	c, err := dgraphtest.NewLocalCluster(conf)
+	require.NoError(t, c.Start())
+
+	defer func() { c.Cleanup(t.Failed()) }()
+
+	gc, cleanup, err := c.Client()
+	require.NoError(t, err)
+	defer cleanup()
+
+	require.NoError(t, gc.LoginIntoNamespace(context.Background(),
+		dgraphapi.DefaultUser, dgraphapi.DefaultPassword, x.RootNamespace))
+
+	require.NoError(t, gc.DropAll())
+	require.NoError(t, gc.SetupSchema(testSchemaWithoutIndex))
+
+	numVectors := 1000
+	rdfs, vectors := dgraphapi.GenerateRandomVectors(0, numVectors, 100, pred)
+	mu := &api.Mutation{SetNquads: []byte(rdfs), CommitNow: true}
+	_, err = gc.Mutate(mu)
+	require.NoError(t, err)
+	require.NoError(t, gc.SetupSchema(vsuite.schema))
+
+	query := `{
+		vector(func: has(project_description_v)) {
+			   count(uid)
+			}
+	}`
+
+	result, err := gc.Query(query)
+	require.NoError(t, err)
+	require.JSONEq(t, fmt.Sprintf(`{"vector":[{"count":%v}]}`, numVectors), string(result.GetJson()))
+
+	for _, vect := range vectors {
+		similarVects, err := gc.QueryMultipleVectorsUsingSimilarTo(vect, pred, 100)
+		require.NoError(t, err)
+		require.Equal(t, 100, len(similarVects))
+	}
+}
+
 func (vsuite *VectorTestSuite) TestVectorIndexWithoutSchemaWithoutIndex() {
 	t := vsuite.T()
 	conf := dgraphtest.NewClusterConfig().WithNumAlphas(1).WithNumZeros(1).WithReplicas(1).WithACL(time.Hour)

tok/kmeans/kmeans.go

@@ -25,6 +25,7 @@ type Kmeans[T c.Float] struct {
 func CreateKMeans[T c.Float](floatBits int, pred string, distFunc func(a, b []T, floatBits int) (T, error)) index.VectorPartitionStrat[T] {
 	return &Kmeans[T]{
 		floatBits: floatBits,
+		numPasses: 5,
 		centroids: &vectorCentroids[T]{
 			distFunc:  distFunc,
 			floatBits: floatBits,
@@ -46,6 +47,9 @@ func (km *Kmeans[T]) GetCentroids() [][]T {
 }
 
 func (km *Kmeans[T]) FindIndexForSearch(vec []T) ([]int, error) {
+	if km.NumPasses() == 0 {
+		return []int{0}, nil
+	}
 	res := make([]int, km.NumSeedVectors())
 	for i := range res {
 		res[i] = i

tok/partitioned_hnsw/partitioned_hnsw.go

@@ -39,7 +39,7 @@ type partitionedHNSW[T c.Float] struct {
 
 func (ph *partitionedHNSW[T]) applyOptions(o opt.Options) error {
 	ph.numClusters, _, _ = opt.GetOpt(o, NumClustersOpt, 1000)
-	ph.vectorDimension, _, _ = opt.GetOpt(o, vectorDimension, 0)
+	ph.vectorDimension, _, _ = opt.GetOpt(o, vectorDimension, -1)
 	ph.partitionStrat, _, _ = opt.GetOpt(o, PartitionStratOpt, "kmeans")
 
 	if ph.partitionStrat != "kmeans" && ph.partitionStrat != "query" {