WIP

internal/db/collection.go

@@ -71,7 +71,7 @@ func (c *collection) newFetcher() fetcher.Fetcher {
 	if c.fetcherFactory != nil {
 		innerFetcher = c.fetcherFactory()
 	} else {
-		innerFetcher = new(fetcher.DocumentFetcher)
+		innerFetcher = fetcher.NewDocumentFetcher()
 	}
 
 	return lens.NewFetcher(innerFetcher, c.db.LensRegistry())

internal/db/fetcher/deleted.go

@@ -0,0 +1,106 @@
+// Copyright 2024 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 fetcher
+
+import (
+	"errors"
+
+	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/immutable"
+)
+
+type deleted struct {
+	activeFetcher fetcher
+	activeDocID   immutable.Option[string]
+
+	deletedFetcher fetcher
+	deletedDocID   immutable.Option[string]
+
+	currentFetcher fetcher
+}
+
+var _ fetcher = (*deleted)(nil)
+
+func newDeletedFetcher(
+	activeFetcher fetcher,
+	deletedFetcher fetcher,
+) *deleted {
+	return &deleted{
+		activeFetcher:  activeFetcher,
+		deletedFetcher: deletedFetcher,
+	}
+}
+
+func (f *deleted) NextDoc() (immutable.Option[string], error) {
+	if !f.activeDocID.HasValue() {
+		var err error
+		f.activeDocID, err = f.activeFetcher.NextDoc()
+		if err != nil {
+			return immutable.None[string](), err
+		}
+	}
+
+	if !f.deletedDocID.HasValue() {
+		var err error
+		f.deletedDocID, err = f.deletedFetcher.NextDoc()
+		if err != nil {
+			return immutable.None[string](), err
+		}
+	}
+
+	if !f.activeDocID.HasValue() || (f.deletedDocID.HasValue() && f.deletedDocID.Value() < f.activeDocID.Value()) {
+		f.currentFetcher = f.deletedFetcher
+		return f.deletedDocID, nil
+	}
+
+	f.currentFetcher = f.activeFetcher
+	return f.activeDocID, nil
+}
+
+func (f *deleted) GetFields(fields ...client.FieldID) (immutable.Option[EncodedDocument], error) {
+	doc, err := f.currentFetcher.GetFields(fields...)
+	if err != nil {
+		return immutable.None[EncodedDocument](), err
+	}
+
+	if f.activeFetcher == f.currentFetcher {
+		f.activeDocID = immutable.None[string]()
+	} else {
+		f.deletedDocID = immutable.None[string]()
+	}
+
+	return doc, nil
+}
+
+func (f *deleted) ExecInfo() ExecInfo {
+	activeExecInfo := f.activeFetcher.ExecInfo()
+	deletedExecInfo := f.deletedFetcher.ExecInfo()
+
+	result := ExecInfo{}
+	result.Add(activeExecInfo)
+	result.Add(deletedExecInfo)
+
+	return result
+}
+
+func (f *deleted) Close() error {
+	activeErr := f.activeFetcher.Close()
+	if activeErr != nil {
+		deletedErr := f.deletedFetcher.Close()
+		if deletedErr != nil {
+			return errors.Join(activeErr, deletedErr)
+		}
+
+		return activeErr
+	}
+
+	return f.deletedFetcher.Close()
+}

internal/db/fetcher/document.go

@@ -0,0 +1,181 @@
+// Copyright 2024 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 fetcher
+
+import (
+	"bytes"
+	"context"
+
+	dsq "github.com/ipfs/go-datastore/query"
+	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/defradb/datastore/iterable"
+	"github.com/sourcenetwork/defradb/internal/db/base"
+	"github.com/sourcenetwork/defradb/internal/keys"
+	"github.com/sourcenetwork/immutable"
+)
+
+type document struct {
+	col        client.Collection
+	fieldsByID map[uint32]client.FieldDefinition
+
+	kvResultsIter dsq.Results
+	currentKV     immutable.Option[keyValue]
+	nextKV        immutable.Option[keyValue]
+	status        client.DocumentStatus
+
+	execInfo ExecInfo
+}
+
+var _ fetcher = (*document)(nil)
+
+func newDocumentFetcher(
+	ctx context.Context,
+	col client.Collection,
+	fields []client.FieldDefinition, // todo - rip out (you parameterized this (that was for ACP - might need both - consider ripping out the GetFields param))
+	kvIter iterable.Iterator,
+	prefix *keys.DataStoreKey,
+	status client.DocumentStatus,
+) (*document, error) {
+	if len(fields) == 0 {
+		fields = col.Definition().GetFields()
+	}
+
+	fieldsByID := make(map[uint32]client.FieldDefinition, len(fields))
+	for _, field := range fields {
+		fieldsByID[uint32(field.ID)] = field
+	}
+
+	if status == client.Active {
+		p := prefix.WithValueFlag()
+		prefix = &p
+	} else if status == client.Deleted {
+		p := prefix.WithDeletedFlag()
+		prefix = &p
+	}
+
+	kvResultsIter, err := kvIter.IteratePrefix(ctx, prefix.ToDS(), prefix.PrefixEnd().ToDS())
+	if err != nil {
+		return nil, err
+	}
+
+	return &document{
+		col:           col,
+		fieldsByID:    fieldsByID,
+		kvResultsIter: kvResultsIter,
+		status:        status,
+	}, nil
+}
+
+func (f *document) NextDoc() (immutable.Option[string], error) {
+	f.execInfo.Reset() // todo - revisit call stack and the adding
+
+	if f.nextKV.HasValue() {
+		docID := f.nextKV.Value().Key.DocID
+		f.currentKV = f.nextKV
+		f.nextKV = immutable.None[keyValue]()
+		return immutable.Some(docID), nil
+	}
+
+	res, ok := f.kvResultsIter.NextSync()
+	if res.Error != nil {
+		return immutable.None[string](), res.Error
+	}
+	if !ok {
+		return immutable.None[string](), nil
+	}
+
+	dsKey, err := keys.NewDataStoreKey(res.Key)
+	if err != nil {
+		return immutable.None[string](), err
+	}
+
+	f.currentKV = immutable.Some(keyValue{
+		Key:   dsKey,
+		Value: res.Value,
+	})
+
+	f.execInfo.DocsFetched++
+
+	return immutable.Some(dsKey.DocID), nil
+}
+
+func (f *document) GetFields(fields ...client.FieldID) (immutable.Option[EncodedDocument], error) {
+	results := make([]keyValue, 0, len(f.col.Schema().Fields))
+	results = append(results, f.currentKV.Value())
+
+	for {
+		res, ok := f.kvResultsIter.NextSync()
+		if !ok {
+			break
+		}
+
+		dsKey, err := keys.NewDataStoreKey(res.Key)
+		if err != nil {
+			return immutable.None[EncodedDocument](), err
+		}
+
+		kv := keyValue{
+			Key:   dsKey,
+			Value: res.Value,
+		}
+
+		if dsKey.DocID != f.currentKV.Value().Key.DocID {
+			f.nextKV = immutable.Some(kv)
+			break
+		}
+
+		results = append(results, kv)
+	}
+
+	encodedDoc := encodedDocument{}
+	encodedDoc.id = []byte(results[0].Key.DocID)
+	encodedDoc.status = f.status
+	encodedDoc.properties = make(map[client.FieldDefinition]*encProperty, len(results))
+
+	for _, r := range results {
+		if r.Key.FieldID == keys.DATASTORE_DOC_VERSION_FIELD_ID {
+			encodedDoc.schemaVersionID = string(r.Value)
+			continue
+		}
+
+		// we have to skip the object marker
+		if bytes.Equal(r.Value, []byte{base.ObjectMarker}) {
+			continue
+		}
+
+		fieldID, err := r.Key.FieldIDAsUint()
+		if err != nil {
+			return immutable.None[EncodedDocument](), err
+		}
+
+		fieldDesc, ok := f.fieldsByID[fieldID]
+		if !ok {
+			continue
+		}
+
+		f.execInfo.FieldsFetched++
+
+		encodedDoc.properties[fieldDesc] = &encProperty{
+			Desc: fieldDesc,
+			Raw:  r.Value,
+		}
+	}
+
+	return immutable.Some[EncodedDocument](&encodedDoc), nil
+}
+
+func (f *document) ExecInfo() ExecInfo {
+	return f.execInfo
+}
+
+func (f *document) Close() error {
+	return f.kvResultsIter.Close()
+}

internal/db/fetcher/fetcher.go

@@ -77,6 +77,13 @@ type Fetcher interface {
 	Close() error
 }
 
+type fetcher interface {
+	NextDoc() (immutable.Option[string], error)
+	GetFields(fields ...client.FieldID) (immutable.Option[EncodedDocument], error)
+	ExecInfo() ExecInfo
+	Close() error
+}
+
 // keyValue is a KV store response containing the resulting core.Key and byte array value.
 type keyValue struct {
 	Key   keys.DataStoreKey

internal/db/fetcher/filtered.go

@@ -0,0 +1,79 @@
+// Copyright 2024 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 fetcher
+
+import (
+	"github.com/sourcenetwork/immutable"
+
+	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/defradb/internal/core"
+	"github.com/sourcenetwork/defradb/internal/planner/mapper"
+)
+
+type filtered struct {
+	filter  *mapper.Filter
+	mapping *core.DocumentMapping
+
+	fetcher fetcher
+}
+
+var _ fetcher = (*filtered)(nil)
+
+func newFilteredFetcher(
+	filter *mapper.Filter,
+	mapping *core.DocumentMapping,
+	fetcher fetcher,
+) *filtered {
+	return &filtered{
+		filter:  filter,
+		mapping: mapping,
+		fetcher: fetcher,
+	}
+}
+
+func (f *filtered) NextDoc() (immutable.Option[string], error) {
+	return f.fetcher.NextDoc()
+}
+
+func (f *filtered) GetFields(fields ...client.FieldID) (immutable.Option[EncodedDocument], error) {
+	doc, err := f.fetcher.GetFields(fields...)
+	if err != nil {
+		return immutable.None[EncodedDocument](), err
+	}
+
+	if !doc.HasValue() {
+		return immutable.None[EncodedDocument](), nil
+	}
+
+	decodedDoc, err := DecodeToDoc(doc.Value(), f.mapping, false)
+	if err != nil {
+		return immutable.None[EncodedDocument](), err
+	}
+
+	passedFilter, err := mapper.RunFilter(decodedDoc, f.filter)
+	if err != nil {
+		return immutable.None[EncodedDocument](), err
+	}
+
+	if !passedFilter {
+		return immutable.None[EncodedDocument](), nil
+	}
+
+	return doc, nil
+}
+
+func (f *filtered) ExecInfo() ExecInfo {
+	return f.fetcher.ExecInfo()
+}
+
+func (f *filtered) Close() error {
+	return f.fetcher.Close()
+}