feat: Non-unique secondary index

.github/codecov.yml

@@ -62,5 +62,6 @@ comment:
 
 ignore:
   - "tests"
+  - "**/mocks/*"
   - "**/*_test.go"
   - "**/*.pb.go"

client/collection.go

@@ -136,6 +136,19 @@ type Collection interface {
 
 	// GetAllDocKeys returns all the document keys that exist in the collection.
 	GetAllDocKeys(ctx context.Context) (<-chan DocKeysResult, error)
+
+	// CreateIndex creates a new index on the collection.
+	// `IndexDescription` contains the description of the index to be created.
+	// `IndexDescription.Name` must start with a letter or an underscore and can
+	// only contain letters, numbers, and underscores.
+	// If the name of the index is not provided, it will be generated.
+	CreateIndex(context.Context, IndexDescription) (IndexDescription, error)
+
+	// DropIndex drops an index from the collection.
+	DropIndex(ctx context.Context, indexName string) error
+
+	// GetIndexes returns all the indexes that exist on the collection.
+	GetIndexes(ctx context.Context) ([]IndexDescription, error)
 }
 
 // DocKeysResult wraps the result of an attempt at a DocKey retrieval operation.

client/descriptions.go

@@ -29,6 +29,9 @@ type CollectionDescription struct {
 
 	// Schema contains the data type information that this Collection uses.
 	Schema SchemaDescription
+
+	// Indexes contains the secondary indexes that this Collection has.
+	Indexes []IndexDescription
 }
 
 // IDString returns the collection ID as a string.
@@ -50,10 +53,10 @@ func (col CollectionDescription) GetField(name string) (FieldDescription, bool)
 
 // GetFieldByID searches for a field with the given ID. If such a field is found it
 // will return it and true, if it is not found it will return false.
-func (col CollectionDescription) GetFieldByID(id string) (FieldDescription, bool) {
+func (col CollectionDescription) GetFieldByID(id FieldID) (FieldDescription, bool) {
 	if !col.Schema.IsEmpty() {
 		for _, field := range col.Schema.Fields {
-			if field.ID.String() == id {
+			if field.ID == id {
 				return field, true
 			}
 		}

client/index.go

@@ -0,0 +1,39 @@
+// 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 client
+
+// IndexDirection is the direction of an index.
+type IndexDirection string
+
+const (
+	// Ascending is the value to use for an ascending fields
+	Ascending IndexDirection = "ASC"
+	// Descending is the value to use for an descending fields
+	Descending IndexDirection = "DESC"
+)
+
+// IndexFieldDescription describes how a field is being indexed.
+type IndexedFieldDescription struct {
+	// Name contains the name of the field.
+	Name string
+	// Direction contains the direction of the index.
+	Direction IndexDirection
+}
+
+// IndexDescription describes an index.
+type IndexDescription struct {
+	// Name contains the name of the index.
+	Name string
+	// ID is the local identifier of this index.
+	ID uint32
+	// Fields contains the fields that are being indexed.
+	Fields []IndexedFieldDescription
+}

core/key.go

@@ -44,6 +44,7 @@ const (
 	COLLECTION                = "/collection/names"
 	COLLECTION_SCHEMA         = "/collection/schema"
 	COLLECTION_SCHEMA_VERSION = "/collection/version"
+	COLLECTION_INDEX          = "/collection/index"
 	SEQ                       = "/seq"
 	PRIMARY_KEY               = "/pk"
 	REPLICATOR                = "/replicator/id"
@@ -67,6 +68,18 @@ type DataStoreKey struct {
 
 var _ Key = (*DataStoreKey)(nil)
 
+// IndexDataStoreKey is key of an indexed document in the database.
+type IndexDataStoreKey struct {
+	// CollectionID is the id of the collection
+	CollectionID uint32
+	// IndexID is the id of the index
+	IndexID uint32
+	// FieldValues is the values of the fields in the index
+	FieldValues [][]byte
+}
+
+var _ Key = (*IndexDataStoreKey)(nil)
+
 type PrimaryDataStoreKey struct {
 	CollectionId string
 	DocKey       string
@@ -106,6 +119,16 @@ type CollectionSchemaVersionKey struct {
 
 var _ Key = (*CollectionSchemaVersionKey)(nil)
 
+// CollectionIndexKey to a stored description of an index
+type CollectionIndexKey struct {
+	// CollectionName is the name of the collection that the index is on
+	CollectionName string
+	// IndexName is the name of the index
+	IndexName string
+}
+
+var _ Key = (*CollectionIndexKey)(nil)
+
 type P2PCollectionKey struct {
 	CollectionID string
 }
@@ -210,6 +233,56 @@ func NewCollectionSchemaVersionKey(schemaVersionId string) CollectionSchemaVersi
 	return CollectionSchemaVersionKey{SchemaVersionId: schemaVersionId}
 }
 
+// NewCollectionIndexKey creates a new CollectionIndexKey from a collection name and index name.
+func NewCollectionIndexKey(colID, indexName string) CollectionIndexKey {
+	return CollectionIndexKey{CollectionName: colID, IndexName: indexName}
+}
+
+// NewCollectionIndexKeyFromString creates a new CollectionIndexKey from a string.
+// It expects the input string is in the following format:
+//
+// /collection/index/[CollectionName]/[IndexName]
+//
+// Where [IndexName] might be omitted. Anything else will return an error.
+func NewCollectionIndexKeyFromString(key string) (CollectionIndexKey, error) {
+	keyArr := strings.Split(key, "/")
+	if len(keyArr) < 4 || len(keyArr) > 5 || keyArr[1] != "collection" || keyArr[2] != "index" {
+		return CollectionIndexKey{}, ErrInvalidKey
+	}
+	result := CollectionIndexKey{CollectionName: keyArr[3]}
+	if len(keyArr) == 5 {
+		result.IndexName = keyArr[4]
+	}
+	return result, nil
+}
+
+// ToString returns the string representation of the key
+// It is in the following format:
+// /collection/index/[CollectionName]/[IndexName]
+// if [CollectionName] is empty, the rest is ignored
+func (k CollectionIndexKey) ToString() string {
+	result := COLLECTION_INDEX
+
+	if k.CollectionName != "" {
+		result = result + "/" + k.CollectionName
+		if k.IndexName != "" {
+			result = result + "/" + k.IndexName
+		}
+	}
+
+	return result
+}
+
+// Bytes returns the byte representation of the key
+func (k CollectionIndexKey) Bytes() []byte {
+	return []byte(k.ToString())
+}
+
+// ToDS returns the datastore key
+func (k CollectionIndexKey) ToDS() ds.Key {
+	return ds.NewKey(k.ToString())
+}
+
 func NewSequenceKey(name string) SequenceKey {
 	return SequenceKey{SequenceName: name}
 }
@@ -318,6 +391,109 @@ func (k DataStoreKey) ToPrimaryDataStoreKey() PrimaryDataStoreKey {
 	}
 }
 
+// NewIndexDataStoreKey creates a new IndexDataStoreKey from a string.
+// It expects the input string is in the following format:
+//
+// /[CollectionID]/[IndexID]/[FieldValue](/[FieldValue]...)
+//
+// Where [CollectionID] and [IndexID] are integers
+func NewIndexDataStoreKey(key string) (IndexDataStoreKey, error) {
+	if key == "" {
+		return IndexDataStoreKey{}, ErrEmptyKey
+	}
+
+	if !strings.HasPrefix(key, "/") {
+		return IndexDataStoreKey{}, ErrInvalidKey
+	}
+
+	elements := strings.Split(key[1:], "/")
+
+	// With less than 3 elements, we know it's an invalid key
+	if len(elements) < 3 {
+		return IndexDataStoreKey{}, ErrInvalidKey
+	}
+
+	colID, err := strconv.Atoi(elements[0])
+	if err != nil {
+		return IndexDataStoreKey{}, ErrInvalidKey
+	}
+
+	indexKey := IndexDataStoreKey{CollectionID: uint32(colID)}
+
+	indID, err := strconv.Atoi(elements[1])
+	if err != nil {
+		return IndexDataStoreKey{}, ErrInvalidKey
+	}
+	indexKey.IndexID = uint32(indID)
+
+	// first 2 elements are the collection and index IDs, the rest are field values
+	for i := 2; i < len(elements); i++ {
+		indexKey.FieldValues = append(indexKey.FieldValues, []byte(elements[i]))
+	}
+
+	return indexKey, nil
+}
+
+// Bytes returns the byte representation of the key
+func (k *IndexDataStoreKey) Bytes() []byte {
+	return []byte(k.ToString())
+}
+
+// ToDS returns the datastore key
+func (k *IndexDataStoreKey) ToDS() ds.Key {
+	return ds.NewKey(k.ToString())
+}
+
+// ToString returns the string representation of the key
+// It is in the following format:
+// /[CollectionID]/[IndexID]/[FieldValue](/[FieldValue]...)
+// If while composing the string from left to right, a component
+// is empty, the string is returned up to that point
+func (k *IndexDataStoreKey) ToString() string {
+	sb := strings.Builder{}
+
+	if k.CollectionID == 0 {
+		return ""
+	}
+	sb.WriteByte('/')
+	sb.WriteString(strconv.Itoa(int(k.CollectionID)))
+
+	if k.IndexID == 0 {
+		return sb.String()
+	}
+	sb.WriteByte('/')
+	sb.WriteString(strconv.Itoa(int(k.IndexID)))
+
+	for _, v := range k.FieldValues {
+		if len(v) == 0 {
+			break
+		}
+		sb.WriteByte('/')
+		sb.WriteString(string(v))
+	}
+
+	return sb.String()
+}
+
+// Equal returns true if the two keys are equal
+func (k IndexDataStoreKey) Equal(other IndexDataStoreKey) bool {
+	if k.CollectionID != other.CollectionID {
+		return false
+	}
+	if k.IndexID != other.IndexID {
+		return false
+	}
+	if len(k.FieldValues) != len(other.FieldValues) {
+		return false
+	}
+	for i := range k.FieldValues {
+		if string(k.FieldValues[i]) != string(other.FieldValues[i]) {
+			return false
+		}
+	}
+	return true
+}
+
 func (k PrimaryDataStoreKey) ToDataStoreKey() DataStoreKey {
 	return DataStoreKey{
 		CollectionID: k.CollectionId,