diff --git a/internal/db/fetcher/deleted.go b/internal/db/fetcher/deleted.go
new file mode 100644
index 0000000000..b408f5ee46
--- /dev/null
+++ b/internal/db/fetcher/deleted.go
@@ -0,0 +1,95 @@
+// 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.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) 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()
+}
diff --git a/internal/db/fetcher/document.go b/internal/db/fetcher/document.go
new file mode 100644
index 0000000000..b8c2c2eb3e
--- /dev/null
+++ b/internal/db/fetcher/document.go
@@ -0,0 +1,150 @@
+// 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 (
+	"context"
+
+	dsq "github.com/ipfs/go-datastore/query"
+	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/defradb/datastore/iterable"
+	"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]
+	status        client.DocumentStatus
+}
+
+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
+	}
+
+	kvResultsIter, err := kvIter.IteratePrefix(ctx, prefix.ToDS(), prefix.PrefixEnd().ToDS())
+	if err != nil {
+		return nil, err
+	}
+
+	return &document{
+		kvResultsIter: kvResultsIter,
+		status:        status,
+	}, nil
+}
+
+func (f *document) NextDoc() (immutable.Option[string], error) {
+	if f.currentKV.HasValue() {
+		docID := f.currentKV.Value().Key.DocID
+		f.currentKV = 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,
+	})
+
+	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())
+
+	res, ok := f.kvResultsIter.NextSync()
+	for ok {
+		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.currentKV = immutable.Some(kv)
+			break
+		}
+
+		results = append(results, kv)
+		res, ok = f.kvResultsIter.NextSync()
+	}
+
+	encodedDoc := encodedDocument{}
+	encodedDoc.id = []byte(results[0].Key.DocID)
+	encodedDoc.status = f.status
+
+	for _, r := range results {
+		if r.Key.FieldID == keys.DATASTORE_DOC_VERSION_FIELD_ID {
+			encodedDoc.schemaVersionID = string(r.Value)
+			continue
+		}
+
+		fieldID, err := r.Key.FieldIDAsUint()
+		if err != nil {
+			return immutable.None[EncodedDocument](), err
+		}
+
+		fieldDesc, ok := f.fieldsByID[fieldID]
+		if !ok {
+			return immutable.None[EncodedDocument](), nil
+		}
+
+		encodedDoc.properties[fieldDesc] = &encProperty{
+			Desc: fieldDesc,
+			Raw:  r.Value,
+		}
+	}
+
+	return immutable.Some[EncodedDocument](&encodedDoc), nil
+}
+
+func (f *document) Close() error {
+	return f.kvResultsIter.Close()
+}
diff --git a/internal/db/fetcher/fetcher.go b/internal/db/fetcher/fetcher.go
index 877cbfa7c8..5abaa7c54c 100644
--- a/internal/db/fetcher/fetcher.go
+++ b/internal/db/fetcher/fetcher.go
@@ -77,6 +77,12 @@ type Fetcher interface {
 	Close() error
 }
 
+type fetcher interface {
+	NextDoc() (immutable.Option[string], error)
+	GetFields(fields ...client.FieldID) (immutable.Option[EncodedDocument], error)
+	Close() error
+}
+
 // keyValue is a KV store response containing the resulting core.Key and byte array value.
 type keyValue struct {
 	Key   keys.DataStoreKey
diff --git a/internal/db/fetcher/filtered.go b/internal/db/fetcher/filtered.go
new file mode 100644
index 0000000000..1a8bd7baf4
--- /dev/null
+++ b/internal/db/fetcher/filtered.go
@@ -0,0 +1,75 @@
+// 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) Close() error {
+	return f.fetcher.Close()
+}
diff --git a/internal/db/fetcher/permissioned.go b/internal/db/fetcher/permissioned.go
new file mode 100644
index 0000000000..f61ed858b1
--- /dev/null
+++ b/internal/db/fetcher/permissioned.go
@@ -0,0 +1,87 @@
+// 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 (
+	"context"
+
+	"github.com/sourcenetwork/immutable"
+
+	"github.com/sourcenetwork/defradb/acp"
+	acpIdentity "github.com/sourcenetwork/defradb/acp/identity"
+	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/defradb/internal/db/permission"
+)
+
+type permissioned struct {
+	ctx context.Context
+
+	identity immutable.Option[acpIdentity.Identity]
+	acp      acp.ACP
+	col      client.Collection
+
+	fetcher fetcher
+}
+
+var _ fetcher = (*permissioned)(nil)
+
+func newPermissionedFetcher(
+	ctx context.Context,
+	identity immutable.Option[acpIdentity.Identity],
+	acp acp.ACP,
+	col client.Collection,
+	fetcher fetcher,
+) *permissioned {
+	return &permissioned{
+		ctx:      ctx,
+		identity: identity,
+		acp:      acp,
+		col:      col,
+		fetcher:  fetcher,
+	}
+}
+
+func (f *permissioned) NextDoc() (immutable.Option[string], error) {
+	docID, err := f.fetcher.NextDoc()
+	if err != nil {
+		return immutable.None[string](), err
+	}
+
+	if !docID.HasValue() {
+		return immutable.None[string](), nil
+	}
+
+	hasPermission, err := permission.CheckAccessOfDocOnCollectionWithACP(
+		f.ctx,
+		f.identity,
+		f.acp,
+		f.col,
+		acp.ReadPermission,
+		docID.Value(),
+	)
+	if err != nil {
+		return immutable.None[string](), err
+	}
+
+	if !hasPermission {
+		return f.NextDoc()
+	}
+
+	return docID, nil
+}
+
+func (f *permissioned) GetFields(fields ...client.FieldID) (immutable.Option[EncodedDocument], error) {
+	return f.fetcher.GetFields(fields...)
+}
+
+func (f *permissioned) Close() error {
+	return f.fetcher.Close()
+}
diff --git a/internal/db/fetcher/prefix.go b/internal/db/fetcher/prefix.go
new file mode 100644
index 0000000000..1edb1ee558
--- /dev/null
+++ b/internal/db/fetcher/prefix.go
@@ -0,0 +1,99 @@
+// 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 (
+	"context"
+
+	dsq "github.com/ipfs/go-datastore/query"
+	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/defradb/datastore"
+	"github.com/sourcenetwork/defradb/datastore/iterable"
+	"github.com/sourcenetwork/defradb/internal/keys"
+	"github.com/sourcenetwork/immutable"
+)
+
+type prefix struct {
+	kvIter iterable.Iterator
+
+	prefixes      []*keys.DataStoreKey
+	currentPrefix int
+
+	fetcher fetcher
+
+	ctx    context.Context
+	col    client.Collection
+	fields []client.FieldDefinition
+	status client.DocumentStatus
+}
+
+var _ fetcher = (*prefix)(nil)
+
+func newPrefixFetcher(
+	ctx context.Context,
+	txn datastore.Txn,
+	prefixes []*keys.DataStoreKey,
+	col client.Collection,
+	fields []client.FieldDefinition,
+	status client.DocumentStatus,
+) (*prefix, error) {
+	kvIter, err := txn.Datastore().GetIterator(dsq.Query{
+		Orders: []dsq.Order{dsq.OrderByKey{}},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &prefix{
+		kvIter:        kvIter,
+		prefixes:      prefixes,
+		currentPrefix: -1,
+		ctx:           ctx,
+		col:           col,
+		fields:        fields,
+		status:        status,
+	}, nil
+}
+
+func (f *prefix) NextDoc() (immutable.Option[string], error) {
+	docID, err := f.fetcher.NextDoc()
+	if err != nil {
+		return immutable.None[string](), err
+	}
+
+	if !docID.HasValue() {
+		if len(f.prefixes) > f.currentPrefix {
+			if f.fetcher != nil {
+				f.fetcher.Close()
+			}
+			f.currentPrefix++
+
+			prefix := f.prefixes[f.currentPrefix]
+
+			f.fetcher, err = newDocumentFetcher(f.ctx, f.col, f.fields, f.kvIter, prefix, f.status)
+			if err != nil {
+				return immutable.None[string](), err
+			}
+
+			return f.NextDoc()
+		}
+	}
+
+	return docID, nil
+}
+
+func (f *prefix) GetFields(fields ...client.FieldID) (immutable.Option[EncodedDocument], error) {
+	return f.fetcher.GetFields(fields...)
+}
+
+func (f *prefix) Close() error {
+	return f.kvIter.Close()
+}
diff --git a/internal/db/fetcher/wrapper.go b/internal/db/fetcher/wrapper.go
new file mode 100644
index 0000000000..d7a230bfa5
--- /dev/null
+++ b/internal/db/fetcher/wrapper.go
@@ -0,0 +1,132 @@
+// 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 (
+	"context"
+
+	"github.com/sourcenetwork/immutable"
+
+	"github.com/sourcenetwork/defradb/acp"
+	acpIdentity "github.com/sourcenetwork/defradb/acp/identity"
+	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/defradb/datastore"
+	"github.com/sourcenetwork/defradb/internal/core"
+	"github.com/sourcenetwork/defradb/internal/keys"
+	"github.com/sourcenetwork/defradb/internal/planner/mapper"
+)
+
+type wrapper struct {
+	// Init props:
+	identity    immutable.Option[acpIdentity.Identity]
+	txn         datastore.Txn
+	acp         immutable.Option[acp.ACP]
+	col         client.Collection
+	fields      []client.FieldDefinition
+	filter      *mapper.Filter
+	docMapper   *core.DocumentMapping
+	showDeleted bool
+
+	fetcher fetcher
+}
+
+var _ Fetcher = (*wrapper)(nil)
+
+func NewDocumentFetcher() Fetcher {
+	return &wrapper{}
+}
+
+func (f *wrapper) Init(
+	ctx context.Context,
+	identity immutable.Option[acpIdentity.Identity],
+	txn datastore.Txn,
+	acp immutable.Option[acp.ACP],
+	col client.Collection,
+	fields []client.FieldDefinition,
+	filter *mapper.Filter,
+	docMapper *core.DocumentMapping,
+	showDeleted bool,
+) error {
+	f.identity = identity
+	f.txn = txn
+	f.acp = acp
+	f.col = col
+	f.fields = fields
+	f.filter = filter
+	f.docMapper = docMapper
+	f.showDeleted = showDeleted
+
+	return nil
+}
+
+func (f *wrapper) Start(ctx context.Context, prefixes ...keys.Walkable) error {
+	dsPrefixes := make([]*keys.DataStoreKey, 0, len(prefixes))
+	for _, prefix := range prefixes {
+		dsPrefix, ok := prefix.(*keys.DataStoreKey)
+		if !ok {
+			continue
+		}
+
+		dsPrefixes = append(dsPrefixes, dsPrefix)
+	}
+
+	var fetcher fetcher
+	fetcher, err := newPrefixFetcher(ctx, f.txn, dsPrefixes, f.col, f.fields, client.Active)
+	if err != nil {
+		return nil
+	}
+
+	if f.showDeleted {
+		deletedFetcher, err := newPrefixFetcher(ctx, f.txn, dsPrefixes, f.col, f.fields, client.Deleted)
+		if err != nil {
+			return nil
+		}
+
+		fetcher = newDeletedFetcher(fetcher, deletedFetcher)
+	}
+
+	if f.acp.HasValue() {
+		fetcher = newPermissionedFetcher(ctx, f.identity, f.acp.Value(), f.col, fetcher)
+	}
+
+	if f.filter != nil {
+		fetcher = newFilteredFetcher(f.filter, f.docMapper, fetcher)
+	}
+
+	f.fetcher = fetcher
+	return nil
+}
+
+func (f *wrapper) FetchNext(ctx context.Context) (EncodedDocument, ExecInfo, error) {
+	docID, err := f.fetcher.NextDoc()
+	if err != nil {
+		return nil, ExecInfo{}, err
+	}
+
+	if !docID.HasValue() {
+		return nil, ExecInfo{}, nil
+	}
+
+	doc, err := f.fetcher.GetFields()
+	if err != nil {
+		return nil, ExecInfo{}, err
+	}
+
+	if !doc.HasValue() {
+		return nil, ExecInfo{}, nil
+	}
+
+	return doc.Value(), ExecInfo{}, nil
+}
+
+func (f *wrapper) Close() error {
+	return f.fetcher.Close()
+}