Repository: genjidb/genji
Pull request #334 (2020-12-30 11:32:53)

Add new stream package
This PR adds a new `stream` package who's gonna be used as the low level query system.
It will replace the existing stream type in future iterations.


Patch:
diff --git a/document/document.go b/document/document.go
index e3b2cc3f..8dfb7c87 100644
--- a/document/document.go
+++ b/document/document.go
@@ -434,6 +434,23 @@ func (fb *FieldBuffer) Fields() []string {
 // A Path represents the path to a particular value within a document.
 type Path []PathFragment
 
+// NewPath creates a path from a list of strings representing either a field name
+// or an array index in string form.
+func NewPath(fragments ...string) Path {
+	var path Path
+
+	for _, frag := range fragments {
+		idx, err := strconv.Atoi(frag)
+		if err != nil {
+			path = append(path, PathFragment{FieldName: frag})
+		} else {
+			path = append(path, PathFragment{ArrayIndex: idx})
+		}
+	}
+
+	return path
+}
+
 // PathFragment is a fragment of a path representing either a field name or
 // the index of an array.
 type PathFragment struct {
diff --git a/go.mod b/go.mod
index c261995b..be4be253 100644
--- a/go.mod
+++ b/go.mod
@@ -4,7 +4,6 @@ go 1.15
 
 require (
 	github.com/buger/jsonparser v1.0.0
-	github.com/davecgh/go-spew v1.1.1
 	github.com/google/btree v1.0.0
 	github.com/stretchr/testify v1.6.1
 	github.com/vmihailenco/msgpack/v5 v5.0.0-beta.1
diff --git a/sql/query/expr/env.go b/sql/query/expr/env.go
index b17fdf6e..4399ec31 100644
--- a/sql/query/expr/env.go
+++ b/sql/query/expr/env.go
@@ -7,14 +7,14 @@ import (
 )
 
 const (
-	currentValueKey = "$v"
+	currentValueKey = "_v"
 )
 
 // Environment contains information about the context in which
 // the expression is evaluated.
 type Environment struct {
 	Params []Param
-	fb     *document.FieldBuffer
+	Buf    *document.FieldBuffer
 
 	Outer *Environment
 }
@@ -25,38 +25,46 @@ func NewEnvironment(v document.Value, params ...Param) *Environment {
 	}
 
 	if v.Type != 0 {
-		env.fb = document.NewFieldBuffer()
+		env.Buf = document.NewFieldBuffer()
 		env.Set(currentValueKey, v)
 	}
 
 	return &env
 }
 
-func (e *Environment) Get(name string) (v document.Value, ok bool) {
-	if e.fb != nil {
-		v, err := e.fb.GetByField(name)
+func (e *Environment) Get(path document.Path) (v document.Value, ok bool) {
+	if e.Buf != nil {
+		v, err := path.GetValueFromDocument(e.Buf)
 		if err == nil {
 			return v, true
 		}
+
+		v, err = e.Buf.GetByField(currentValueKey)
+		if err == nil {
+			v, err = path.GetValue(v)
+			if err == nil {
+				return v, true
+			}
+		}
 	}
 
 	if e.Outer != nil {
-		return e.Outer.Get(name)
+		return e.Outer.Get(path)
 	}
 
 	return
 }
 
 func (e *Environment) Set(name string, v document.Value) {
-	if e.fb == nil {
-		e.fb = document.NewFieldBuffer()
+	if e.Buf == nil {
+		e.Buf = document.NewFieldBuffer()
 	}
 
-	e.fb.Set(document.Path{document.PathFragment{FieldName: name}}, v)
+	e.Buf.Set(document.Path{document.PathFragment{FieldName: name}}, v)
 }
 
 func (e *Environment) GetCurrentValue() (document.Value, bool) {
-	return e.Get(currentValueKey)
+	return e.Get(document.Path{document.PathFragment{FieldName: currentValueKey}})
 }
 
 func (e *Environment) SetCurrentValue(v document.Value) {
@@ -93,3 +101,25 @@ func (e *Environment) GetParamByIndex(pos int) (document.Value, error) {
 
 	return document.NewValue(e.Params[idx].Value)
 }
+
+func (e *Environment) Clone() (*Environment, error) {
+	newEnv := Environment{
+		Params: e.Params,
+		Buf:    document.NewFieldBuffer(),
+	}
+
+	err := newEnv.Buf.Copy(e.Buf)
+	if err != nil {
+		return nil, err
+	}
+
+	if e.Outer != nil {
+		newOuter, err := e.Outer.Clone()
+		if err != nil {
+			return nil, err
+		}
+		newEnv.Outer = newOuter
+	}
+
+	return &newEnv, nil
+}
diff --git a/sql/query/expr/path.go b/sql/query/expr/path.go
index 9176ed8b..20f6f8fb 100644
--- a/sql/query/expr/path.go
+++ b/sql/query/expr/path.go
@@ -1,6 +1,8 @@
 package expr
 
 import (
+	"errors"
+
 	"github.com/genjidb/genji/document"
 )
 
@@ -10,17 +12,20 @@ type Path document.Path
 // Eval extracts the document from the context and selects the right value.
 // It implements the Expr interface.
 func (p Path) Eval(env *Environment) (document.Value, error) {
-	v, ok := env.GetCurrentValue()
-	if !ok {
+	if len(p) == 0 {
+		return nullLitteral, nil
+	}
+
+	if _, ok := env.GetCurrentValue(); !ok {
 		return nullLitteral, document.ErrFieldNotFound
 	}
 
-	v, err := document.Path(p).GetValue(v)
-	if err == document.ErrFieldNotFound || err == document.ErrValueNotFound {
+	v, ok := env.Get(document.Path(p))
+	if !ok {
 		return nullLitteral, nil
 	}
 
-	return v, err
+	return v, nil
 }
 
 // IsEqual compares this expression with the other expression and returns
@@ -41,3 +46,14 @@ func (p Path) IsEqual(other Expr) bool {
 func (p Path) String() string {
 	return document.Path(p).String()
 }
+
+// A Wildcard is an expression that iterates over all the fields of a document.
+type Wildcard struct{}
+
+func (w Wildcard) String() string {
+	return "*"
+}
+
+func (w Wildcard) Eval(env *Environment) (document.Value, error) {
+	return document.Value{}, errors.New("no table specified")
+}
diff --git a/stream/iterator.go b/stream/iterator.go
new file mode 100644
index 00000000..c0dd057b
--- /dev/null
+++ b/stream/iterator.go
@@ -0,0 +1,61 @@
+package stream
+
+import (
+	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/sql/query/expr"
+)
+
+// An Iterator can iterate over values.
+type Iterator interface {
+	// Iterate goes through all the values and calls the given function by passing each one of them.
+	// If the given function returns an error, the iteration stops.
+	Iterate(fn func(env *expr.Environment) error) error
+}
+
+// The IteratorFunc type is an adapter to allow the use of ordinary functions as Iterators.
+// If f is a function with the appropriate signature, IteratorFunc(f) is an Iterator that calls f.
+type IteratorFunc func(fn func(env *expr.Environment) error) error
+
+// Iterate calls f(fn).
+func (f IteratorFunc) Iterate(fn func(env *expr.Environment) error) error {
+	return f(fn)
+}
+
+type valueIterator []document.Value
+
+func (it valueIterator) Iterate(fn func(env *expr.Environment) error) error {
+	var env expr.Environment
+
+	for _, v := range it {
+		env.SetCurrentValue(v)
+		err := fn(&env)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// NewValueIterator creates an iterator that iterates over the given values.
+func NewValueIterator(values ...document.Value) Iterator {
+	return valueIterator(values)
+}
+
+type arrayIterator struct {
+	arr document.Array
+}
+
+func (it *arrayIterator) Iterate(fn func(env *expr.Environment) error) error {
+	var env expr.Environment
+
+	return it.arr.Iterate(func(i int, value document.Value) error {
+		env.SetCurrentValue(value)
+		return fn(&env)
+	})
+}
+
+// NewArrayIterator creats an iterator that iterates over each values of the given array.
+func NewArrayIterator(a document.Array) Iterator {
+	return &arrayIterator{arr: a}
+}
diff --git a/stream/operator.go b/stream/operator.go
new file mode 100644
index 00000000..25a83006
--- /dev/null
+++ b/stream/operator.go
@@ -0,0 +1,465 @@
+package stream
+
+import (
+	"bytes"
+	"container/heap"
+	"fmt"
+
+	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/sql/query/expr"
+)
+
+const (
+	groupEnvKey = "_group"
+	accEnvKey   = "_acc"
+)
+
+// An Operator is used to modify a stream.
+// It takes an environment containing the current value as well as any other metadata
+// created by other operatorsand returns a new environment which will be passed to the next operator.
+// If it returns a nil environment, the env will be ignored.
+// If it returns an error, the stream will be interrupted and that error will bubble up
+// and returned by this function, unless that error is ErrStreamClosed, in which case
+// the Iterate method will stop the iteration and return nil.
+// Stream operators can be reused, and thus, any state or side effect should be kept within the Op closure
+// unless the nature of the operator prevents that.
+type Operator interface {
+	Op() (OperatorFunc, error)
+}
+
+// An OperatorFunc is the function that will receive each value of the stream.
+type OperatorFunc func(env *expr.Environment) (*expr.Environment, error)
+
+// A MapOperator applies an expression on each value of the stream and returns a new value.
+type MapOperator struct {
+	E expr.Expr
+}
+
+// Map evaluates e on each value of the stream and outputs the result.
+func Map(e expr.Expr) *MapOperator {
+	return &MapOperator{E: e}
+}
+
+// Op implements the Operator interface.
+func (m *MapOperator) Op() (OperatorFunc, error) {
+	var newEnv expr.Environment
+
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		v, err := m.E.Eval(env)
+		if err != nil {
+			return nil, err
+		}
+
+		newEnv.SetCurrentValue(v)
+		newEnv.Outer = env
+		return &newEnv, nil
+	}, nil
+}
+
+func (m *MapOperator) String() string {
+	return fmt.Sprintf("map(%s)", m.E)
+}
+
+// A FilterOperator filters values based on a given expression.
+type FilterOperator struct {
+	E expr.Expr
+}
+
+// Filter evaluates e for each incoming value and filters any value whose result is not truthy.
+func Filter(e expr.Expr) *FilterOperator {
+	return &FilterOperator{E: e}
+}
+
+// Op implements the Operator interface.
+func (m *FilterOperator) Op() (OperatorFunc, error) {
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		v, err := m.E.Eval(env)
+		if err != nil {
+			return nil, err
+		}
+
+		ok, err := v.IsTruthy()
+		if err != nil {
+			return nil, err
+		}
+
+		if !ok {
+			return nil, nil
+		}
+
+		return env, nil
+	}, nil
+}
+
+func (m *FilterOperator) String() string {
+	return fmt.Sprintf("filter(%s)", m.E)
+}
+
+// A TakeOperator closes the stream after a certain number of values.
+type TakeOperator struct {
+	E expr.Expr
+}
+
+// Take closes the stream after n values have passed through the operator.
+// n must evaluate to a number or to a value that can be converted to an integer.
+func Take(n expr.Expr) *TakeOperator {
+	return &TakeOperator{E: n}
+}
+
+// Op implements the Operator interface.
+func (m *TakeOperator) Op() (OperatorFunc, error) {
+	var n, count int64
+	v, err := m.E.Eval(&expr.Environment{})
+	if err != nil {
+		return nil, err
+	}
+	if v.Type != document.IntegerValue {
+		v, err = v.CastAsInteger()
+		if err != nil {
+			return nil, err
+		}
+	}
+	n = v.V.(int64)
+
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		if count < n {
+			count++
+			return env, nil
+		}
+
+		return nil, ErrStreamClosed
+	}, nil
+}
+
+func (m *TakeOperator) String() string {
+	return fmt.Sprintf("take(%s)", m.E)
+}
+
+// A SkipOperator skips the n first values of the stream.
+type SkipOperator struct {
+	E expr.Expr
+}
+
+// Skip ignores the first n values of the stream.
+// n must evaluate to a number or to a value that can be converted to an integer.
+func Skip(n expr.Expr) *SkipOperator {
+	return &SkipOperator{E: n}
+}
+
+// Op implements the Operator interface.
+func (m *SkipOperator) Op() (OperatorFunc, error) {
+	var n, skipped int64
+	v, err := m.E.Eval(&expr.Environment{})
+	if err != nil {
+		return nil, err
+	}
+	if v.Type != document.IntegerValue {
+		v, err = v.CastAsInteger()
+		if err != nil {
+			return nil, err
+		}
+	}
+	n = v.V.(int64)
+
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		if skipped < n {
+			skipped++
+			return nil, nil
+		}
+
+		return env, nil
+	}, nil
+}
+
+func (m *SkipOperator) String() string {
+	return fmt.Sprintf("skip(%s)", m.E)
+}
+
+// A GroupByOperator applies an expression on each value of the stream and stores the result in the _group
+// variable in the output stream.
+type GroupByOperator struct {
+	E expr.Expr
+}
+
+// GroupBy applies e on each value of the stream and stores the result in the _group
+// variable in the output stream.
+func GroupBy(e expr.Expr) *GroupByOperator {
+	return &GroupByOperator{E: e}
+}
+
+// Op implements the Operator interface.
+func (op *GroupByOperator) Op() (OperatorFunc, error) {
+	var newEnv expr.Environment
+
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		v, err := op.E.Eval(env)
+		if err != nil {
+			return nil, err
+		}
+
+		newEnv.Set(groupEnvKey, v)
+		newEnv.Outer = env
+		return &newEnv, nil
+	}, nil
+}
+
+func (op *GroupByOperator) String() string {
+	return fmt.Sprintf("groupBy(%s)", op.E)
+}
+
+// A ReduceOperator consumes the given stream and outputs one value per group.
+// It reads the _group variable from the environment to determine with group
+// to assign each value. If no _group variable is available, it will assume all
+// values are part of the same group and reduce them into one value.
+// To reduce incoming values, reduce
+type ReduceOperator struct {
+	Seed, Accumulator expr.Expr
+}
+
+// Reduce consumes the incoming stream and outputs one value per group.
+// It reads the _group variable from the environment to determine whitch group
+// to assign each value. If no _group variable is available, it will assume all
+// values are part of the same group and reduce them into one value.
+// The seed is used to determine the initial value of the reduction. The initial value
+// is stored in the _acc variable of the environment.
+// The accumulator then takes the environment for each incoming value and is used to
+// compute the new value of the _acc variable.
+func Reduce(seed, accumulator expr.Expr) *ReduceOperator {
+	return &ReduceOperator{Seed: seed, Accumulator: accumulator}
+}
+
+// Pipe stores s in the operator and return a new Stream with the reduce operator appended. It implements the Piper interface.
+func (op *ReduceOperator) Pipe(s Stream) Stream {
+	return Stream{
+		it: IteratorFunc(func(fn func(env *expr.Environment) error) error {
+			return op.iterate(s, fn)
+		}),
+	}
+}
+
+// Op implements the Operator interface but should never be called by Stream.
+func (op *ReduceOperator) Op() (OperatorFunc, error) {
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		return env, nil
+	}, nil
+}
+
+func (op *ReduceOperator) iterate(s Stream, fn func(env *expr.Environment) error) error {
+	var b bytes.Buffer
+	enc := document.NewValueEncoder(&b)
+
+	nullValue := document.NewNullValue()
+	err := enc.Encode(nullValue)
+	if err != nil {
+		return err
+	}
+	nullKey := b.String()
+	b.Reset()
+
+	groups := make(map[string]*expr.Environment)
+	var groupKeys []string
+
+	mkGroup := func(outer *expr.Environment, groupValue document.Value, groupKey string) (*expr.Environment, error) {
+		var groupEnv expr.Environment
+		groupEnv.Outer = outer
+		groups[groupKey] = &groupEnv
+		groupKeys = append(groupKeys, groupKey)
+
+		seed, err := op.Seed.Eval(&groupEnv)
+		if err != nil {
+			return nil, err
+		}
+		groupEnv.Set(accEnvKey, seed)
+
+		return &groupEnv, nil
+	}
+
+	err = s.Iterate(func(env *expr.Environment) error {
+		groupValue, ok := env.Get(document.NewPath(groupEnvKey))
+		if !ok {
+			groupValue = nullValue
+		}
+
+		b.Reset()
+		err := enc.Encode(groupValue)
+		if err != nil {
+			return err
+		}
+
+		groupName := b.String()
+		genv, ok := groups[groupName]
+		if !ok {
+			genv, err = mkGroup(env, groupValue, groupName)
+			if err != nil {
+				return err
+			}
+		}
+
+		v, err := op.Accumulator.Eval(genv)
+		if err != nil {
+			return err
+		}
+
+		genv.Set(accEnvKey, v)
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	if len(groups) == 0 {
+		// create one group by default if there was no input
+		mkGroup(nil, nullValue, nullKey)
+	}
+
+	for _, groupKey := range groupKeys {
+		genv := groups[groupKey]
+		acc, _ := genv.Get(document.NewPath(accEnvKey))
+		genv.SetCurrentValue(acc)
+		err = fn(genv)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (op *ReduceOperator) String() string {
+	return fmt.Sprintf("reduce(%s, %s)", op.Seed, op.Accumulator)
+}
+
+// A SortOperator consumes every value of the stream and outputs them in order.
+type SortOperator struct {
+	Expr expr.Expr
+	Desc bool
+}
+
+// Sort consumes every value of the stream and outputs them in order.
+// It operates a partial sort on the iterator using a heap.
+// This ensures a O(k+n log n) time complexity, where k is the sum of
+// Take() + Skip() operators, if provided, otherwise k = n.
+// If the sorting is in ascending order, a min-heap will be used
+// otherwise a max-heap will be used instead.
+// Once the heap is filled entirely with the content of the incoming stream, a stream is returned.
+// During iteration, the stream will pop the k-smallest or k-largest elements, depending on
+// the chosen sorting order (ASC or DESC).
+// This function is not memory efficient as it is loading the entire stream in memory before
+// returning the k-smallest or k-largest elements.
+func Sort(e expr.Expr) *SortOperator {
+	return &SortOperator{Expr: e}
+}
+
+// SortReverse does the same as Sort but in descending order.
+func SortReverse(e expr.Expr) *SortOperator {
+	return &SortOperator{Expr: e, Desc: true}
+}
+
+// Pipe stores s in the operator and return a new Stream with the reduce operator appended. It implements the Piper interface.
+func (op *SortOperator) Pipe(s Stream) Stream {
+	return Stream{
+		it: IteratorFunc(func(fn func(env *expr.Environment) error) error {
+			return op.iterate(s, fn)
+		}),
+	}
+}
+
+// Op implements the Operator interface but should never be called by Stream.
+func (op *SortOperator) Op() (OperatorFunc, error) {
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		return env, nil
+	}, nil
+}
+
+func (op *SortOperator) iterate(s Stream, fn func(env *expr.Environment) error) error {
+	h, err := op.sortStream(s)
+	if err != nil {
+		return err
+	}
+
+	for h.Len() > 0 {
+		node := heap.Pop(h).(heapNode)
+		err := fn(node.data)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (op *SortOperator) sortStream(st Stream) (heap.Interface, error) {
+	var h heap.Interface
+	if op.Desc {
+		h = new(maxHeap)
+	} else {
+		h = new(minHeap)
+	}
+
+	heap.Init(h)
+
+	return h, st.Iterate(func(env *expr.Environment) error {
+		sortV, err := op.Expr.Eval(env)
+		if err != nil {
+			return err
+		}
+
+		// We need to make sure sort behaviour
+		// is the same with or without indexes.
+		// To achieve that, the value must be encoded using the same method
+		// as what the index package would do.
+		var buf bytes.Buffer
+
+		err = document.NewValueEncoder(&buf).Encode(sortV)
+		if err != nil {
+			return err
+		}
+
+		node := heapNode{
+			value: buf.Bytes(),
+		}
+		node.data, err = env.Clone()
+		if err != nil {
+			return err
+		}
+
+		heap.Push(h, node)
+
+		return nil
+	})
+}
+
+func (op *SortOperator) String() string {
+	return fmt.Sprintf("sort(%s)", op.Expr)
+}
+
+type heapNode struct {
+	value []byte
+	data  *expr.Environment
+}
+
+type minHeap []heapNode
+
+func (h minHeap) Len() int           { return len(h) }
+func (h minHeap) Less(i, j int) bool { return bytes.Compare(h[i].value, h[j].value) < 0 }
+func (h minHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
+
+func (h *minHeap) Push(x interface{}) {
+	*h = append(*h, x.(heapNode))
+}
+
+func (h *minHeap) Pop() interface{} {
+	old := *h
+	n := len(old)
+	x := old[n-1]
+	*h = old[0 : n-1]
+	return x
+}
+
+type maxHeap struct {
+	minHeap
+}
+
+func (h maxHeap) Less(i, j int) bool {
+	return bytes.Compare(h.minHeap[i].value, h.minHeap[j].value) > 0
+}
diff --git a/stream/project.go b/stream/project.go
new file mode 100644
index 00000000..4795943f
--- /dev/null
+++ b/stream/project.go
@@ -0,0 +1,127 @@
+package stream
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/sql/query/expr"
+)
+
+// A ProjectOperator applies an expression on each value of the stream and returns a new value.
+type ProjectOperator struct {
+	Exprs []expr.Expr
+}
+
+// Project creates a ProjectOperator.
+func Project(exprs ...expr.Expr) *ProjectOperator {
+	return &ProjectOperator{Exprs: exprs}
+}
+
+// Op implements the Operator interface.
+func (m *ProjectOperator) Op(s Stream) (OperatorFunc, error) {
+	var mask maskDocument
+	v := document.Value{
+		Type: document.DocumentValue,
+		V:    &mask,
+	}
+	var newEnv expr.Environment
+
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		mask.Env = env
+		mask.Exprs = m.Exprs
+		newEnv.SetCurrentValue(v)
+		newEnv.Outer = env
+		return &newEnv, nil
+	}, nil
+}
+
+func (m *ProjectOperator) String() string {
+	var b strings.Builder
+
+	b.WriteString("project(")
+	for i, e := range m.Exprs {
+		b.WriteString(e.(fmt.Stringer).String())
+		if i+1 < len(m.Exprs) {
+			b.WriteString(", ")
+		}
+	}
+	b.WriteString(")")
+	return b.String()
+}
+
+type maskDocument struct {
+	Env   *expr.Environment
+	Exprs []expr.Expr
+}
+
+func (d maskDocument) GetByField(field string) (v document.Value, err error) {
+	for _, e := range d.Exprs {
+		if _, ok := e.(expr.Wildcard); ok {
+			cv, ok := d.Env.GetCurrentValue()
+			if !ok {
+				continue
+			}
+
+			if cv.Type != document.DocumentValue {
+				if cv.String() == field {
+					return cv, nil
+				}
+
+				continue
+			}
+
+			v, err = cv.V.(document.Document).GetByField(field)
+			if err == document.ErrFieldNotFound {
+				continue
+			}
+			return
+		}
+
+		if e.(fmt.Stringer).String() == field {
+			return e.Eval(d.Env)
+		}
+	}
+
+	err = document.ErrFieldNotFound
+	return
+}
+
+func (d maskDocument) Iterate(fn func(field string, value document.Value) error) error {
+	for _, e := range d.Exprs {
+		if _, ok := e.(expr.Wildcard); ok {
+			v, ok := d.Env.GetCurrentValue()
+			if !ok {
+				return nil
+			}
+
+			if v.Type == document.DocumentValue {
+				err := v.V.(document.Document).Iterate(fn)
+				if err != nil {
+					return err
+				}
+
+				continue
+			}
+
+			err := fn(v.String(), v)
+			if err != nil {
+				return err
+			}
+
+			continue
+		}
+
+		v, err := e.Eval(d.Env)
+		if err != nil {
+			return err
+		}
+
+		err = fn(e.(fmt.Stringer).String(), v)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
diff --git a/stream/stream.go b/stream/stream.go
new file mode 100644
index 00000000..ccdc4dbb
--- /dev/null
+++ b/stream/stream.go
@@ -0,0 +1,82 @@
+package stream
+
+import (
+	"errors"
+
+	"github.com/genjidb/genji/sql/query/expr"
+)
+
+// ErrStreamClosed is used to indicate that a stream must be closed.
+var ErrStreamClosed = errors.New("stream closed")
+
+// Stream reads values from an iterator one by one and passes them
+// through a list of operators for transformation.
+type Stream struct {
+	it Iterator
+	op Operator
+}
+
+// New creates a stream using the given iterator.
+func New(it Iterator) Stream {
+	return Stream{
+		it: it,
+	}
+}
+
+// A Piper is an operator that can pipe itself into a stream.
+type Piper interface {
+	Pipe(Stream) Stream
+}
+
+// Pipe creates a new Stream who can read its data from s and apply
+// op to every value passed by its Iterate method.
+// If op implements the Piper interface, it will call its Pipe method instead.
+func (s Stream) Pipe(op Operator) Stream {
+	if p, ok := op.(Piper); ok {
+		return p.Pipe(s)
+	}
+
+	return Stream{
+		it: s,
+		op: op,
+	}
+}
+
+// Iterate calls the underlying iterator's iterate method.
+// If this stream was created using the Pipe method, it will apply fn
+// to any value passed by the underlying iterator.
+// If fn returns an error, the stream will be interrupted and that error will bubble up
+// and returned by fn, unless that error is ErrStreamClosed, in which case
+// the Iterate method will stop the iteration and return nil.
+// It implements the Iterator interface.
+func (s Stream) Iterate(fn func(env *expr.Environment) error) error {
+	if s.it == nil {
+		return nil
+	}
+
+	if s.op == nil {
+		return s.it.Iterate(fn)
+	}
+
+	opFn, err := s.op.Op()
+	if err != nil {
+		return err
+	}
+
+	err = s.it.Iterate(func(env *expr.Environment) error {
+		env, err := opFn(env)
+		if err != nil {
+			return err
+		}
+		if env == nil {
+			return nil
+		}
+
+		return fn(env)
+	})
+	if err != ErrStreamClosed {
+		return err
+	}
+
+	return nil
+}


Test patch:
diff --git a/sql/query/expr/path_test.go b/sql/query/expr/path_test.go
index 50ce4de5..9976c2b4 100644
--- a/sql/query/expr/path_test.go
+++ b/sql/query/expr/path_test.go
@@ -26,6 +26,7 @@ func TestPathExpr(t *testing.T) {
 		}(),
 			false},
 		{"b.`foo bar`[0]", document.NewIntegerValue(1), false},
+		{"_v.b.`foo bar`[0]", document.NewIntegerValue(1), false},
 		{"b.`foo bar`[1]", document.NewIntegerValue(2), false},
 		{"b.`foo bar`[2]", nullLitteral, false},
 		{"b[0]", nullLitteral, false},
diff --git a/stream/operator_test.go b/stream/operator_test.go
new file mode 100644
index 00000000..0ae205e8
--- /dev/null
+++ b/stream/operator_test.go
@@ -0,0 +1,425 @@
+package stream_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/sql/parser"
+	"github.com/genjidb/genji/sql/query/expr"
+	"github.com/genjidb/genji/stream"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMap(t *testing.T) {
+	tests := []struct {
+		e       expr.Expr
+		in, out *expr.Environment
+		fails   bool
+	}{
+		{
+			parser.MustParseExpr("10"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			expr.NewEnvironment(document.NewIntegerValue(10)),
+			false,
+		},
+		{
+			parser.MustParseExpr("null"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			expr.NewEnvironment(document.NewNullValue()),
+			false,
+		},
+		{
+			parser.MustParseExpr("a"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			expr.NewEnvironment(document.NewNullValue()),
+			false,
+		},
+		{
+			parser.MustParseExpr("a"),
+			expr.NewEnvironment(document.NewDocumentValue(document.NewFieldBuffer().Add("a", document.NewIntegerValue(1)))),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%s", test.e), func(t *testing.T) {
+			test.out.Outer = test.in
+
+			op, err := stream.Map(test.e).Op()
+			require.NoError(t, err)
+			env, err := op(test.in)
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.out, env)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, stream.Map(parser.MustParseExpr("1")).String(), "map(1)")
+	})
+}
+
+func TestFilter(t *testing.T) {
+	tests := []struct {
+		e       expr.Expr
+		in, out *expr.Environment
+		fails   bool
+	}{
+		{
+			parser.MustParseExpr("1"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			false,
+		},
+		{
+			parser.MustParseExpr("_v > 1"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			nil,
+			false,
+		},
+		{
+			parser.MustParseExpr("_v >= 1"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			false,
+		},
+		{
+			parser.MustParseExpr("null"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			nil,
+			false,
+		},
+		{
+			parser.MustParseExpr("a"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			nil,
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%s", test.e), func(t *testing.T) {
+			op, err := stream.Filter(test.e).Op()
+			require.NoError(t, err)
+			env, err := op(test.in)
+
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.out, env)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, stream.Filter(parser.MustParseExpr("1")).String(), "filter(1)")
+	})
+}
+
+func TestTake(t *testing.T) {
+	tests := []struct {
+		inNumber int64
+		n        expr.Expr
+		output   int
+		fails    bool
+	}{
+		{5, parser.MustParseExpr("1"), 1, false},
+		{5, parser.MustParseExpr("7"), 5, false},
+		{5, parser.MustParseExpr("1.1"), 1, false},
+		{5, parser.MustParseExpr("true"), 1, false},
+		{5, parser.MustParseExpr("1 + 1"), 2, false},
+		{5, parser.MustParseExpr("a"), 1, true},
+		{5, parser.MustParseExpr("'hello'"), 1, true},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%d/%s", test.inNumber, test.n), func(t *testing.T) {
+			var values []document.Value
+
+			for i := int64(0); i < test.inNumber; i++ {
+				values = append(values, document.NewIntegerValue(i))
+			}
+
+			s := stream.New(stream.NewValueIterator(values...))
+			s = s.Pipe(stream.Take(test.n))
+
+			var count int
+			err := s.Iterate(func(env *expr.Environment) error {
+				count++
+				return nil
+			})
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, count)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, stream.Take(parser.MustParseExpr("1")).String(), "take(1)")
+	})
+}
+
+func TestSkip(t *testing.T) {
+	tests := []struct {
+		inNumber int64
+		n        expr.Expr
+		output   int
+		fails    bool
+	}{
+		{5, parser.MustParseExpr("1"), 4, false},
+		{5, parser.MustParseExpr("7"), 0, false},
+		{5, parser.MustParseExpr("1.1"), 4, false},
+		{5, parser.MustParseExpr("true"), 4, false},
+		{5, parser.MustParseExpr("1 + 1"), 3, false},
+		{5, parser.MustParseExpr("a"), 1, true},
+		{5, parser.MustParseExpr("'hello'"), 1, true},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%d/%s", test.inNumber, test.n), func(t *testing.T) {
+			var values []document.Value
+
+			for i := int64(0); i < test.inNumber; i++ {
+				values = append(values, document.NewIntegerValue(i))
+			}
+
+			s := stream.New(stream.NewValueIterator(values...))
+			s = s.Pipe(stream.Skip(test.n))
+
+			var count int
+			err := s.Iterate(func(env *expr.Environment) error {
+				count++
+				return nil
+			})
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, count)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, stream.Skip(parser.MustParseExpr("1")).String(), "skip(1)")
+	})
+}
+
+func TestGroupBy(t *testing.T) {
+	tests := []struct {
+		e     expr.Expr
+		in    *expr.Environment
+		group document.Value
+		fails bool
+	}{
+		{
+			parser.MustParseExpr("10"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			document.NewIntegerValue(10),
+			false,
+		},
+		{
+			parser.MustParseExpr("null"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			document.NewNullValue(),
+			false,
+		},
+		{
+			parser.MustParseExpr("a"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			document.NewNullValue(),
+			false,
+		},
+		{
+			parser.MustParseExpr("_v"),
+			expr.NewEnvironment(document.NewIntegerValue(1)),
+			document.NewIntegerValue(1),
+			false,
+		},
+		{
+			parser.MustParseExpr("a"),
+			expr.NewEnvironment(document.NewDocumentValue(document.NewFieldBuffer().Add("a", document.NewIntegerValue(1)))),
+			document.NewIntegerValue(1),
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%s", test.e), func(t *testing.T) {
+			var want expr.Environment
+			want.Outer = test.in
+			want.Set("_group", test.group)
+
+			op, err := stream.GroupBy(test.e).Op()
+			require.NoError(t, err)
+			env, err := op(test.in)
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, &want, env)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, stream.GroupBy(parser.MustParseExpr("1")).String(), "groupBy(1)")
+	})
+}
+
+func generateSeqValues(max int64) (values []document.Value) {
+	for i := int64(0); i < max; i++ {
+		values = append(values, document.NewIntegerValue(i))
+	}
+
+	return values
+}
+
+func TestReduce(t *testing.T) {
+	tests := []struct {
+		name      string
+		groupBy   expr.Expr
+		seed, acc expr.Expr
+		values    []document.Value
+		want      []document.Value
+		fails     bool
+	}{
+		{
+			"count",
+			nil,
+			parser.MustParseExpr("0"),
+			parser.MustParseExpr("_acc + 1"),
+			[]document.Value{document.NewIntegerValue(0)},
+			[]document.Value{document.NewIntegerValue(1)},
+			false,
+		},
+		{
+			"count/groupBy",
+			parser.MustParseExpr("_v % 2"),
+			parser.MustParseExpr("0"),
+			parser.MustParseExpr("_acc + 1"),
+			generateSeqValues(10),
+			[]document.Value{document.NewIntegerValue(5), document.NewIntegerValue(5)},
+			false,
+		},
+		{
+			"count/noInput",
+			nil,
+			parser.MustParseExpr("0"),
+			parser.MustParseExpr("_acc + 1"),
+			nil,
+			[]document.Value{document.NewIntegerValue(0)},
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s := stream.New(stream.NewValueIterator(test.values...))
+			if test.groupBy != nil {
+				s = s.Pipe(stream.GroupBy(test.groupBy))
+			}
+			s = s.Pipe(stream.Reduce(test.seed, test.acc))
+
+			var got []document.Value
+			err := s.Iterate(func(env *expr.Environment) error {
+				v, ok := env.GetCurrentValue()
+				require.True(t, ok)
+				got = append(got, v)
+				return nil
+			})
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.want, got)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, `reduce({"count": 0}, {"count": _acc + 1})`, stream.Reduce(parser.MustParseExpr("{count: 0}"), parser.MustParseExpr("{count: _acc +1}")).String())
+	})
+}
+
+func TestSort(t *testing.T) {
+	tests := []struct {
+		name     string
+		sortExpr expr.Expr
+		values   []document.Value
+		want     []document.Value
+		fails    bool
+		desc     bool
+	}{
+		{
+			"ASC",
+			parser.MustParseExpr("_v"),
+			[]document.Value{
+				document.NewIntegerValue(0),
+				document.NewNullValue(),
+				document.NewBoolValue(true),
+			},
+			[]document.Value{
+				document.NewNullValue(),
+				document.NewBoolValue(true),
+				document.NewIntegerValue(0),
+			},
+			false,
+			false,
+		},
+		{
+			"DESC",
+			parser.MustParseExpr("_v"),
+			[]document.Value{
+				document.NewIntegerValue(0),
+				document.NewNullValue(),
+				document.NewBoolValue(true),
+			},
+			[]document.Value{
+				document.NewIntegerValue(0),
+				document.NewBoolValue(true),
+				document.NewNullValue(),
+			},
+			false,
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s := stream.New(stream.NewValueIterator(test.values...))
+			if test.desc {
+				s = s.Pipe(stream.SortReverse(test.sortExpr))
+			} else {
+				s = s.Pipe(stream.Sort(test.sortExpr))
+			}
+
+			var got []document.Value
+			err := s.Iterate(func(env *expr.Environment) error {
+				v, ok := env.GetCurrentValue()
+				require.True(t, ok)
+				got = append(got, v)
+				return nil
+			})
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.want, got)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, `sort(a)`, stream.Sort(parser.MustParseExpr("a")).String())
+	})
+}
diff --git a/stream/project_test.go b/stream/project_test.go
new file mode 100644
index 00000000..151432c7
--- /dev/null
+++ b/stream/project_test.go
@@ -0,0 +1,104 @@
+package stream_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+
+	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/sql/parser"
+	"github.com/genjidb/genji/sql/query/expr"
+	"github.com/genjidb/genji/stream"
+	"github.com/stretchr/testify/require"
+)
+
+func TestProject(t *testing.T) {
+	tests := []struct {
+		name  string
+		exprs []expr.Expr
+		in    document.Value
+		out   string
+		fails bool
+	}{
+		{
+			"Value/Constant",
+			[]expr.Expr{parser.MustParseExpr("10")},
+			document.NewIntegerValue(1),
+			`{"_v":{"10":10}}`,
+			false,
+		},
+		{
+			"Value/Wildcard",
+			[]expr.Expr{expr.Wildcard{}},
+			document.NewIntegerValue(1),
+			`{"_v":{"1":1}}`,
+			false,
+		},
+		{
+			"Value/Multiple",
+			[]expr.Expr{expr.Wildcard{}, expr.Wildcard{}, parser.MustParseExpr("10")},
+			document.NewIntegerValue(1),
+			`{"_v":{"1":1,"1":1,"10":10}}`,
+			false,
+		},
+		{
+			"Document/Constant",
+			[]expr.Expr{parser.MustParseExpr("10")},
+			document.NewDocumentValue(document.NewFromJSON([]byte(`{"a":1,"b":[true]}`))),
+			`{"_v":{"10":10}}`,
+			false,
+		},
+		{
+			"Document/Wildcard",
+			[]expr.Expr{expr.Wildcard{}},
+			document.NewDocumentValue(document.NewFromJSON([]byte(`{"a":1,"b":[true]}`))),
+			`{"_v":{"a":1,"b":[true]}}`,
+			false,
+		},
+		{
+			"Document/Multiple",
+			[]expr.Expr{expr.Wildcard{}, expr.Wildcard{}, parser.MustParseExpr("10")},
+			document.NewDocumentValue(document.NewFromJSON([]byte(`{"a":1,"b":[true]}`))),
+			`{"_v":{"a":1,"b":[true],"a":1,"b":[true],"10":10}}`,
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%s", test.name), func(t *testing.T) {
+			var inEnv expr.Environment
+			inEnv.SetCurrentValue(test.in)
+
+			op, err := stream.Project(test.exprs...).Op(stream.Stream{})
+			require.NoError(t, err)
+			env, err := op(&inEnv)
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, &inEnv, env.Outer)
+				got, _ := json.Marshal(env.Buf)
+				require.Equal(t, test.out, string(got))
+
+				v, _ := env.GetCurrentValue()
+				d := v.V.(document.Document)
+				err = d.Iterate(func(field string, want document.Value) error {
+					got, err := d.GetByField(field)
+					require.NoError(t, err)
+					require.Equal(t, want, got)
+					return nil
+				})
+				require.NoError(t, err)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, "project(1, *, *, 1 + 1)", stream.Project(
+			parser.MustParseExpr("1"),
+			expr.Wildcard{},
+			expr.Wildcard{},
+			parser.MustParseExpr("1 +    1"),
+		).String())
+	})
+}
diff --git a/stream/stream_test.go b/stream/stream_test.go
new file mode 100644
index 00000000..88bd8d33
--- /dev/null
+++ b/stream/stream_test.go
@@ -0,0 +1,32 @@
+package stream_test
+
+import (
+	"testing"
+
+	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/sql/parser"
+	"github.com/genjidb/genji/sql/query/expr"
+	"github.com/genjidb/genji/stream"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStream(t *testing.T) {
+	s := stream.New(stream.NewValueIterator(
+		document.NewIntegerValue(1),
+		document.NewIntegerValue(2),
+	))
+
+	s = s.Pipe(stream.Map(parser.MustParseExpr("_v + 1")))
+	s = s.Pipe(stream.Filter(parser.MustParseExpr("_v > 2")))
+
+	var count int64
+	err := s.Iterate(func(env *expr.Environment) error {
+		v, ok := env.GetCurrentValue()
+		require.True(t, ok)
+		require.Equal(t, document.NewIntegerValue(count+3), v)
+		count++
+		return nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, int64(1), count)
+}