feat: Make queries utilise secondary indexes

client/index.go

@@ -37,3 +37,22 @@ type IndexDescription struct {
 	// Fields contains the fields that are being indexed.
 	Fields []IndexedFieldDescription
 }
+
+// CollectIndexedFields returns all fields that are indexed by all collection indexes.
+func (d CollectionDescription) CollectIndexedFields(schema *SchemaDescription) []FieldDescription {
+	fieldsMap := make(map[string]bool)
+	fields := make([]FieldDescription, 0, len(d.Indexes))
+	for _, index := range d.Indexes {
+		for _, field := range index.Fields {
+			for i := range schema.Fields {
+				colField := schema.Fields[i]
+				if field.Name == colField.Name && !fieldsMap[field.Name] {
+					fieldsMap[field.Name] = true
+					fields = append(fields, colField)
+					break
+				}
+			}
+		}
+	}
+	return fields
+}

datastore/errors.go

@@ -14,6 +14,10 @@ import (
 	"github.com/sourcenetwork/defradb/errors"
 )
 
+const (
+	errInvalidStoredValue string = "invalid stored value"
+)
+
 // Errors returnable from this package.
 //
 // This list is incomplete and undefined errors may also be returned.
@@ -26,3 +30,9 @@ var (
 	// ErrNotFound is an error returned when a block is not found.
 	ErrNotFound = errors.New("blockstore: block not found")
 )
+
+// NewErrInvalidStoredValue returns a new error indicating that the stored
+// value in the database is invalid.
+func NewErrInvalidStoredValue(inner error) error {
+	return errors.Wrap(errInvalidStoredValue, inner)
+}

datastore/prefix_query.go

@@ -0,0 +1,81 @@
+// Copyright 2023 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package datastore
+
+import (
+	"context"
+	"encoding/json"
+
+	ds "github.com/ipfs/go-datastore"
+
+	"github.com/ipfs/go-datastore/query"
+)
+
+// DeserializePrefix deserializes all elements with the given prefix from the given storage.
+// It returns the keys and their corresponding elements.
+func DeserializePrefix[T any](
+	ctx context.Context,
+	prefix string,
+	store ds.Read,
+) ([]string, []T, error) {
+	q, err := store.Query(ctx, query.Query{Prefix: prefix})
+	if err != nil {
+		return nil, nil, err
+	}
+
+	keys := make([]string, 0)
+	elements := make([]T, 0)
+	for res := range q.Next() {
+		if res.Error != nil {
+			_ = q.Close()
+			return nil, nil, res.Error
+		}
+
+		var element T
+		err = json.Unmarshal(res.Value, &element)
+		if err != nil {
+			_ = q.Close()
+			return nil, nil, NewErrInvalidStoredValue(err)
+		}
+		keys = append(keys, res.Key)
+		elements = append(elements, element)
+	}
+	if err := q.Close(); err != nil {
+		return nil, nil, err
+	}
+	return keys, elements, nil
+}
+
+// FetchKeysForPrefix fetches all keys with the given prefix from the given storage.
+func FetchKeysForPrefix(
+	ctx context.Context,
+	prefix string,
+	store ds.Read,
+) ([]ds.Key, error) {
+	q, err := store.Query(ctx, query.Query{Prefix: prefix})
+	if err != nil {
+		return nil, err
+	}
+
+	keys := make([]ds.Key, 0)
+	for res := range q.Next() {
+		if res.Error != nil {
+			_ = q.Close()
+			return nil, res.Error
+		}
+		keys = append(keys, ds.NewKey(res.Key))
+	}
+	if err = q.Close(); err != nil {
+		return nil, err
+	}
+
+	return keys, nil
+}

db/collection_index.go

@@ -17,9 +17,6 @@ import (
 	"strconv"
 	"strings"
 
-	ds "github.com/ipfs/go-datastore"
-	"github.com/ipfs/go-datastore/query"
-
 	"github.com/sourcenetwork/defradb/client"
 	"github.com/sourcenetwork/defradb/core"
 	"github.com/sourcenetwork/defradb/datastore"
@@ -63,7 +60,7 @@ func (db *db) getAllIndexes(
 ) (map[client.CollectionName][]client.IndexDescription, error) {
 	prefix := core.NewCollectionIndexKey("", "")
 
-	deserializedIndexes, err := deserializePrefix[client.IndexDescription](ctx,
+	keys, indexDescriptions, err := datastore.DeserializePrefix[client.IndexDescription](ctx,
 		prefix.ToString(), txn.Systemstore())
 
 	if err != nil {
@@ -72,12 +69,15 @@ func (db *db) getAllIndexes(
 
 	indexes := make(map[client.CollectionName][]client.IndexDescription)
 
-	for _, indexRec := range deserializedIndexes {
-		indexKey, err := core.NewCollectionIndexKeyFromString(indexRec.key)
+	for i := range keys {
+		indexKey, err := core.NewCollectionIndexKeyFromString(keys[i])
 		if err != nil {
 			return nil, NewErrInvalidStoredIndexKey(indexKey.ToString())
 		}
-		indexes[indexKey.CollectionName] = append(indexes[indexKey.CollectionName], indexRec.element)
+		indexes[indexKey.CollectionName] = append(
+			indexes[indexKey.CollectionName],
+			indexDescriptions[i],
+		)
 	}
 
 	return indexes, nil
@@ -89,16 +89,12 @@ func (db *db) fetchCollectionIndexDescriptions(
 	colName string,
 ) ([]client.IndexDescription, error) {
 	prefix := core.NewCollectionIndexKey(colName, "")
-	deserializedIndexes, err := deserializePrefix[client.IndexDescription](ctx,
+	_, indexDescriptions, err := datastore.DeserializePrefix[client.IndexDescription](ctx,
 		prefix.ToString(), txn.Systemstore())
 	if err != nil {
 		return nil, err
 	}
-	indexes := make([]client.IndexDescription, 0, len(deserializedIndexes))
-	for _, indexRec := range deserializedIndexes {
-		indexes = append(indexes, indexRec.element)
-	}
-	return indexes, nil
+	return indexDescriptions, nil
 }
 
 func (c *collection) indexNewDoc(ctx context.Context, txn datastore.Txn, doc *client.Document) error {
@@ -115,27 +111,6 @@ func (c *collection) indexNewDoc(ctx context.Context, txn datastore.Txn, doc *cl
 	return nil
 }
 
-// collectIndexedFields returns all fields that are indexed by all collection indexes.
-func (c *collection) collectIndexedFields() []client.FieldDescription {
-	fieldsMap := make(map[string]client.FieldDescription)
-	for _, index := range c.indexes {
-		for _, field := range index.Description().Fields {
-			for i := range c.desc.Schema.Fields {
-				colField := c.desc.Schema.Fields[i]
-				if field.Name == colField.Name {
-					fieldsMap[field.Name] = colField
-					break
-				}
-			}
-		}
-	}
-	fields := make([]client.FieldDescription, 0, len(fieldsMap))
-	for _, field := range fieldsMap {
-		fields = append(fields, field)
-	}
-	return fields
-}
-
 func (c *collection) updateIndexedDoc(
 	ctx context.Context,
 	txn datastore.Txn,
@@ -145,7 +120,13 @@ func (c *collection) updateIndexedDoc(
 	if err != nil {
 		return err
 	}
-	oldDoc, err := c.get(ctx, txn, c.getPrimaryKeyFromDocKey(doc.Key()), c.collectIndexedFields(), false)
+	desc := c.Description()
+	oldDoc, err := c.get(
+		ctx,
+		txn,
+		c.getPrimaryKeyFromDocKey(doc.Key()), desc.CollectIndexedFields(&desc.Schema),
+		false,
+	)
 	if err != nil {
 		return err
 	}
@@ -370,7 +351,7 @@ func (c *collection) dropIndex(ctx context.Context, txn datastore.Txn, indexName
 func (c *collection) dropAllIndexes(ctx context.Context, txn datastore.Txn) error {
 	prefix := core.NewCollectionIndexKey(c.Name(), "")
 
-	keys, err := fetchKeysForPrefix(ctx, prefix.ToString(), txn.Systemstore())
+	keys, err := datastore.FetchKeysForPrefix(ctx, prefix.ToString(), txn.Systemstore())
 	if err != nil {
 		return err
 	}
@@ -510,39 +491,3 @@ func generateIndexName(col client.Collection, fields []client.IndexedFieldDescri
 	}
 	return sb.String()
 }
-
-type deserializedElement[T any] struct {
-	key     string
-	element T
-}
-
-func deserializePrefix[T any](
-	ctx context.Context,
-	prefix string,
-	storage ds.Read,
-) ([]deserializedElement[T], error) {
-	q, err := storage.Query(ctx, query.Query{Prefix: prefix})
-	if err != nil {
-		return nil, NewErrFailedToCreateCollectionQuery(err)
-	}
-
-	elements := make([]deserializedElement[T], 0)
-	for res := range q.Next() {
-		if res.Error != nil {
-			_ = q.Close()
-			return nil, res.Error
-		}
-
-		var element T
-		err = json.Unmarshal(res.Value, &element)
-		if err != nil {
-			_ = q.Close()
-			return nil, NewErrInvalidStoredIndex(err)
-		}
-		elements = append(elements, deserializedElement[T]{key: res.Key, element: element})
-	}
-	if err := q.Close(); err != nil {
-		return nil, err
-	}
-	return elements, nil
-}

db/fetcher/encoded_doc.go

@@ -131,6 +131,24 @@
 	return doc, nil
 }
 
+// MergeProperties merges the properties of the given document into this document.
+// Existing fields of the current document are overwritten.
+func (encdoc *encodedDocument) MergeProperties(other EncodedDocument) {
+	otherEncDoc, ok := other.(*encodedDocument)
+	if !ok {
+		return
+	}
+	for field, prop := range otherEncDoc.properties {
+		encdoc.properties[field] = prop
+	}
+	if other.Key() != nil {
+		encdoc.key = other.Key()
+	}
+	if other.SchemaVersionID() != "" {
+		encdoc.schemaVersionID = other.SchemaVersionID()
+	}
+}
+
 // DecodeToDoc returns a decoded document as a
 // map of field/value pairs
 func DecodeToDoc(encdoc EncodedDocument, mapping *core.DocumentMapping, filter bool) (core.Doc, error) {

db/fetcher/fetcher.go

@@ -33,18 +33,22 @@ type ExecInfo struct {
 	DocsFetched uint64
 	// Number of fields fetched.
 	FieldsFetched uint64
+	// Number of indexes fetched.
+	IndexesFetched uint64
 }
 
 // Add adds the other ExecInfo to the current ExecInfo.
 func (s *ExecInfo) Add(other ExecInfo) {
 	s.DocsFetched += other.DocsFetched
 	s.FieldsFetched += other.FieldsFetched
+	s.IndexesFetched += other.IndexesFetched
 }
 
 // Reset resets the ExecInfo.
 func (s *ExecInfo) Reset() {
 	s.DocsFetched = 0
 	s.FieldsFetched = 0
+	s.IndexesFetched = 0
 }
 
 // Fetcher is the interface for collecting documents from the underlying data store.
@@ -576,6 +580,8 @@ func (df *DocumentFetcher) fetchNext(ctx context.Context) (EncodedDocument, Exec
 	// keyparts := df.kv.Key.List()
 	// key := keyparts[len(keyparts)-2]
 
+	prevExecInfo := df.execInfo
+	defer func() { df.execInfo.Add(prevExecInfo) }()
 	df.execInfo.Reset()
 	// iterate until we have collected all the necessary kv pairs for the doc
 	// we'll know when were done when either