Repository: genjidb/genji
Pull request #337 (2021-01-05 05:47:59)

Add io stream operators
This PR adds some stream operators that interact with indexes and tables.

codecov-io: # [Codecov](https://codecov.io/gh/genjidb/genji/pull/337?src=pr&el=h1) Report
> Merging [#337](https://codecov.io/gh/genjidb/genji/pull/337?src=pr&el=desc) (a3e3123) into [main](https://codecov.io/gh/genjidb/genji/commit/7f63c1bd2fbfc00d7e7a51366855e3e611593bff?el=desc) (7f63c1b) will **increase** coverage by `0.14%`.
> The diff coverage is `68.02%`.

[![Impacted file tree graph](https://codecov.io/gh/genjidb/genji/pull/337/graphs/tree.svg?width=650&height=150&src=pr&token=nwR1GpvtyH)](https://codecov.io/gh/genjidb/genji/pull/337?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##             main     #337      +/-   ##
==========================================
+ Coverage   63.30%   63.44%   +0.14%     
==========================================
  Files          80       80              
  Lines        7229     7491     +262     
==========================================
+ Hits         4576     4753     +177     
- Misses       2053     2112      +59     
- Partials      600      626      +26     
```


| [Impacted Files](https://codecov.io/gh/genjidb/genji/pull/337?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [document/create.go](https://codecov.io/gh/genjidb/genji/pull/337/diff?src=pr&el=tree#diff-ZG9jdW1lbnQvY3JlYXRlLmdv) | `67.37% <0.00%> (-0.49%)` | :arrow_down: |
| [document/document.go](https://codecov.io/gh/genjidb/genji/pull/337/diff?src=pr&el=tree#diff-ZG9jdW1lbnQvZG9jdW1lbnQuZ28=) | `68.44% <0.00%> (-0.79%)` | :arrow_down: |
| [database/table.go](https://codecov.io/gh/genjidb/genji/pull/337/diff?src=pr&el=tree#diff-ZGF0YWJhc2UvdGFibGUuZ28=) | `46.80% <33.33%> (-3.68%)` | :arrow_down: |
| [stream/operator.go](https://codecov.io/gh/genjidb/genji/pull/337/diff?src=pr&el=tree#diff-c3RyZWFtL29wZXJhdG9yLmdv) | `73.12% <64.10%> (-4.03%)` | :arrow_down: |
| [stream/iterator.go](https://codecov.io/gh/genjidb/genji/pull/337/diff?src=pr&el=tree#diff-c3RyZWFtL2l0ZXJhdG9yLmdv) | `83.44% <82.48%> (-16.56%)` | :arrow_down: |
| [document/value.go](https://codecov.io/gh/genjidb/genji/pull/337/diff?src=pr&el=tree#diff-ZG9jdW1lbnQvdmFsdWUuZ28=) | `73.59% <100.00%> (+0.07%)` | :arrow_up: |
| [sql/query/insert.go](https://codecov.io/gh/genjidb/genji/pull/337/diff?src=pr&el=tree#diff-c3FsL3F1ZXJ5L2luc2VydC5nbw==) | `72.72% <100.00%> (+1.29%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/genjidb/genji/pull/337?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/genjidb/genji/pull/337?src=pr&el=footer). Last update [7f63c1b...a3e3123](https://codecov.io/gh/genjidb/genji/pull/337?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/database/table.go b/database/table.go
index 7b19ccc1..94375733 100644
--- a/database/table.go
+++ b/database/table.go
@@ -44,7 +44,8 @@ func (t *Table) Truncate() error {
 // If a primary key has been specified during the table creation, the field is expected to be present
 // in the given document.
 // If no primary key has been selected, a monotonic autoincremented integer key will be generated.
-func (t *Table) Insert(d document.Document) ([]byte, error) {
+// It returns the inserted document alongside its key. They key can be accessed using the document.Keyer interface.
+func (t *Table) Insert(d document.Document) (document.Document, error) {
 	info, err := t.Info()
 	if err != nil {
 		return nil, err
@@ -103,7 +104,16 @@ func (t *Table) Insert(d document.Document) ([]byte, error) {
 		}
 	}
 
-	return key, nil
+	if fb, ok := d.(*document.FieldBuffer); ok {
+		fb.EncodedKey = key
+		return fb, nil
+	}
+
+	return documentWithKey{
+		Document: d,
+		key:      key,
+		pk:       info.GetPrimaryKey(),
+	}, nil
 }
 
 // Delete a document by key.
@@ -270,18 +280,18 @@ func (t *Table) Indexes() (map[string]Index, error) {
 	return indexes, nil
 }
 
-type encodedDocumentWithKey struct {
+type documentWithKey struct {
 	document.Document
 
 	key []byte
 	pk  *FieldConstraint
 }
 
-func (e encodedDocumentWithKey) RawKey() []byte {
+func (e documentWithKey) RawKey() []byte {
 	return e.key
 }
 
-func (e encodedDocumentWithKey) Key() (document.Value, error) {
+func (e documentWithKey) Key() (document.Value, error) {
 	if e.pk == nil {
 		docid, _ := binary.Uvarint(e.key)
 		return document.NewIntegerValue(int64(docid)), nil
@@ -354,22 +364,106 @@ func (d *lazilyDecodedDocument) copyFromItem() error {
 // Iterate goes through all the documents of the table and calls the given function by passing each one of them.
 // If the given function returns an error, the iteration stops.
 func (t *Table) Iterate(fn func(d document.Document) error) error {
-	// To avoid unnecessary allocations, we create the struct once and reuse
-	// it during each iteration.
-	d := lazilyDecodedDocument{
-		codec: t.tx.db.Codec,
+	return t.AscendGreaterOrEqual(document.Value{}, fn)
+}
+
+// EncodeValueToKey encodes a value following primary key constraints.
+// It returns a binary representation of the key as used in the store.
+// It can be used to manually add a new entry to the store or to compare
+// with other keys during table iteration.
+func (t *Table) EncodeValueToKey(v document.Value) ([]byte, error) {
+	info, err := t.Info()
+	if err != nil {
+		return nil, err
+	}
+
+	return t.encodeValueToKey(info, v)
+}
+
+func (t *Table) encodeValueToKey(info *TableInfo, v document.Value) ([]byte, error) {
+	var err error
+
+	pk := info.GetPrimaryKey()
+	if pk == nil {
+		// if no primary key was defined, convert the pivot to an integer then to an unsigned integer
+		// and encode it as a varint
+		v, err = v.CastAsInteger()
+		if err != nil {
+			return nil, err
+		}
+
+		docid := uint64(v.V.(int64))
+
+		buf := make([]byte, binary.MaxVarintLen64)
+		n := binary.PutUvarint(buf, docid)
+		return buf[:n], nil
 	}
 
+	// if a primary key was defined and the primary is typed, convert the value to the right type.
+	if !pk.Type.IsZero() {
+		v, err = v.CastAs(pk.Type)
+		if err != nil {
+			return nil, err
+		}
+
+		return v.MarshalBinary()
+	}
+
+	// it no primary key type is specified,
+	// encode keys regardless of type.
+	var buf bytes.Buffer
+	err = document.NewValueEncoder(&buf).Encode(v)
+	if err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+// AscendGreaterOrEqual iterates over the documents of the table whose key
+// is greater than or equal to the pivot.
+// The pivot is converted to the type of the primary key, if any, prior to iteration.
+// If the pivot is empty, it iterates from the beginning of the table.
+func (t *Table) AscendGreaterOrEqual(pivot document.Value, fn func(d document.Document) error) error {
+	return t.iterate(pivot, false, fn)
+}
+
+// DescendLessOrEqual iterates over the documents of the table whose key
+// is less than or equal to the pivot, in reverse order.
+// The pivot is converted to the type of the primary key, if any, prior to iteration.
+// If the pivot is empty, it iterates from the end of the table in reverse order.
+func (t *Table) DescendLessOrEqual(pivot document.Value, fn func(d document.Document) error) error {
+	return t.iterate(pivot, true, fn)
+}
+
+func (t *Table) iterate(pivot document.Value, reverse bool, fn func(d document.Document) error) error {
+	var seek []byte
+
 	info, err := t.Info()
 	if err != nil {
 		return err
 	}
+
+	// if there is a pivot, convert it to the right type
+	if !pivot.Type.IsZero() {
+		var err error
+		seek, err = t.encodeValueToKey(info, pivot)
+		if err != nil {
+			return err
+		}
+	}
+
+	// To avoid unnecessary allocations, we create the struct once and reuse
+	// it during each iteration.
+	d := lazilyDecodedDocument{
+		codec: t.tx.db.Codec,
+	}
+
 	d.pk = info.GetPrimaryKey()
 
-	it := t.Store.Iterator(engine.IteratorOptions{})
+	it := t.Store.Iterator(engine.IteratorOptions{Reverse: reverse})
 	defer it.Close()
 
-	for it.Seek(nil); it.Valid(); it.Next() {
+	for it.Seek(seek); it.Valid(); it.Next() {
 		d.Reset()
 		d.item = it.Item()
 		// d must be passed as pointer, not value,
@@ -402,7 +496,7 @@ func (t *Table) GetDocument(key []byte) (document.Document, error) {
 		return nil, err
 	}
 
-	var d encodedDocumentWithKey
+	var d documentWithKey
 	d.Document = t.tx.db.Codec.NewDocument(v)
 	d.key = key
 	d.pk = info.GetPrimaryKey()
diff --git a/document/create.go b/document/create.go
index f80047f6..e606de9e 100644
--- a/document/create.go
+++ b/document/create.go
@@ -48,6 +48,10 @@ func (j jsonEncodedDocument) GetByField(field string) (Value, error) {
 	return parseJSONValue(dt, v)
 }
 
+func (j jsonEncodedDocument) MarshalJSON() ([]byte, error) {
+	return j.data, nil
+}
+
 // NewFromMap creates a document from a map.
 // Due to the way maps are designed, iteration order is not guaranteed.
 func NewFromMap(m interface{}) (Document, error) {
diff --git a/document/document.go b/document/document.go
index 8dfb7c87..338de78e 100644
--- a/document/document.go
+++ b/document/document.go
@@ -344,6 +344,14 @@ func (fb *FieldBuffer) Copy(d Document) error {
 	return nil
 }
 
+// Clone the buffer.
+func (fb *FieldBuffer) Clone() *FieldBuffer {
+	var newFb FieldBuffer
+
+	_ = newFb.Copy(fb)
+	return &newFb
+}
+
 // Apply a function to all the values of the buffer.
 func (fb *FieldBuffer) Apply(fn func(p Path, v Value) (Value, error)) error {
 	path := Path{PathFragment{}}
diff --git a/document/value.go b/document/value.go
index edae7eae..9fe38424 100644
--- a/document/value.go
+++ b/document/value.go
@@ -90,6 +90,11 @@ func (t ValueType) IsNumber() bool {
 	return t == IntegerValue || t == DoubleValue
 }
 
+// IsZero returns whether this is a valid type.
+func (t ValueType) IsZero() bool {
+	return t == 0
+}
+
 // A Value stores encoded data alongside its type.
 type Value struct {
 	Type ValueType
diff --git a/sql/query/insert.go b/sql/query/insert.go
index bf236e16..7014d4db 100644
--- a/sql/query/insert.go
+++ b/sql/query/insert.go
@@ -63,10 +63,11 @@ func (stmt InsertStmt) insertDocuments(t *database.Table, env *expr.Environment)
 			return res, fmt.Errorf("expected document, got %s", v.Type)
 		}
 
-		res.LastInsertKey, err = t.Insert(v.V.(document.Document))
+		d, err := t.Insert(v.V.(document.Document))
 		if err != nil {
 			return res, err
 		}
+		res.LastInsertKey = d.(document.Keyer).RawKey()
 
 		res.RowsAffected++
 	}
@@ -103,11 +104,13 @@ func (stmt InsertStmt) insertExprList(t *database.Table, env *expr.Environment)
 			return nil
 		})
 
-		res.LastInsertKey, err = t.Insert(&fb)
+		d, err := t.Insert(&fb)
 		if err != nil {
 			return res, err
 		}
 
+		res.LastInsertKey = d.(document.Keyer).RawKey()
+
 		res.RowsAffected++
 	}
 
diff --git a/stream/iterator.go b/stream/iterator.go
index 17a0555d..7b18ed44 100644
--- a/stream/iterator.go
+++ b/stream/iterator.go
@@ -1,6 +1,11 @@
 package stream
 
 import (
+	"bytes"
+	"errors"
+	"fmt"
+
+	"github.com/genjidb/genji/database"
 	"github.com/genjidb/genji/document"
 	"github.com/genjidb/genji/sql/query/expr"
 )
@@ -41,3 +46,292 @@ func (it documentIterator) Iterate(fn func(env *expr.Environment) error) error {
 func NewDocumentIterator(documents ...document.Document) Iterator {
 	return documentIterator(documents)
 }
+
+// A TableIterator iterates over the documents of a table.
+type TableIterator struct {
+	Name    string
+	Table   *database.Table
+	Params  []expr.Param
+	Min     document.Value
+	Max     document.Value
+	Reverse bool
+}
+
+// NewTableIterator creates an iterator that iterates over each document of the given table.
+func NewTableIterator(name string) *TableIterator {
+	return &TableIterator{Name: name}
+}
+
+// TableIteratorOptions are used to control the iteration range and direction.
+type TableIteratorOptions struct {
+	Min     document.Value
+	Max     document.Value
+	Reverse bool
+}
+
+// NewTableIteratorWithOptions creates an iterator that iterates over each document of the given table.
+func NewTableIteratorWithOptions(name string, opt TableIteratorOptions) *TableIterator {
+	return &TableIterator{
+		Name:    name,
+		Min:     opt.Min,
+		Max:     opt.Max,
+		Reverse: opt.Reverse,
+	}
+}
+
+// Bind the iterator to the table and parameters.
+func (it *TableIterator) Bind(tx *database.Transaction, params []expr.Param) error {
+	var err error
+
+	it.Table, err = tx.GetTable(it.Name)
+	if err != nil {
+		return err
+	}
+	it.Params = params
+	return nil
+}
+
+// Iterate over the documents of the table. Each document is stored in the environment
+// that is passed to the fn function, using SetCurrentValue.
+func (it *TableIterator) Iterate(fn func(env *expr.Environment) error) error {
+	var env expr.Environment
+	env.Params = it.Params
+
+	var min, max []byte
+	var err error
+
+	if !it.Min.Type.IsZero() {
+		min, err = it.Table.EncodeValueToKey(it.Min)
+		if err != nil {
+			return err
+		}
+	}
+
+	if !it.Max.Type.IsZero() {
+		max, err = it.Table.EncodeValueToKey(it.Max)
+		if err != nil {
+			return err
+		}
+	}
+
+	errStop := errors.New("stop")
+
+	if !it.Reverse {
+		if max == nil {
+			return it.Table.AscendGreaterOrEqual(it.Min, func(d document.Document) error {
+				env.SetDocument(d)
+				return fn(&env)
+			})
+		}
+		err := it.Table.AscendGreaterOrEqual(it.Min, func(d document.Document) error {
+			k := d.(document.Keyer).RawKey()
+
+			// if there is an upper bound, iterate until we reach the max key
+			if bytes.Compare(k, max) >= 0 {
+				return errStop
+			}
+
+			env.SetDocument(d)
+			return fn(&env)
+		})
+		if err == errStop {
+			err = nil
+		}
+		return err
+	}
+
+	if min == nil {
+		return it.Table.DescendLessOrEqual(it.Max, func(d document.Document) error {
+			env.SetDocument(d)
+			return fn(&env)
+		})
+	}
+
+	err = it.Table.DescendLessOrEqual(it.Max, func(d document.Document) error {
+		k := d.(document.Keyer).RawKey()
+
+		// if there is a lower bound, iterate until we reach the min key
+		if bytes.Compare(k, min) <= 0 {
+			return errStop
+		}
+
+		env.SetDocument(d)
+		return fn(&env)
+	})
+	if err == errStop {
+		err = nil
+	}
+	return err
+}
+
+func (it *TableIterator) String() string {
+	var min, max, reverse string
+	if !it.Min.Type.IsZero() {
+		min = it.Min.String()
+	}
+	if !it.Max.Type.IsZero() {
+		max = it.Max.String()
+	}
+
+	reverse = "+"
+	if it.Reverse {
+		reverse = "-"
+	}
+
+	return fmt.Sprintf("%s%s[%s:%s]", reverse, it.Name, min, max)
+}
+
+// A IndexIterator iterates over the documents of an index.
+type IndexIterator struct {
+	Name    string
+	Index   *database.Index
+	Table   *database.Table
+	Params  []expr.Param
+	Min     document.Value
+	Max     document.Value
+	Reverse bool
+}
+
+// Bind the iterator to the index and parameters.
+func (it *IndexIterator) Bind(tx *database.Transaction, params []expr.Param) error {
+	var err error
+
+	it.Index, err = tx.GetIndex(it.Name)
+	if err != nil {
+		return err
+	}
+
+	it.Table, err = tx.GetTable(it.Index.Opts.TableName)
+	if err != nil {
+		return err
+	}
+	it.Params = params
+	return nil
+}
+
+// NewIndexIterator creates an iterator that iterates over each document of the given table.
+func NewIndexIterator(name string) *IndexIterator {
+	return &IndexIterator{Name: name}
+}
+
+// IndexIteratorOptions are used to control the iteration range and direction.
+type IndexIteratorOptions struct {
+	Min     document.Value
+	Max     document.Value
+	Reverse bool
+}
+
+// NewIndexIteratorWithOptions creates an iterator that iterates over each document of the given table.
+func NewIndexIteratorWithOptions(name string, opt IndexIteratorOptions) *IndexIterator {
+	return &IndexIterator{
+		Name:    name,
+		Min:     opt.Min,
+		Max:     opt.Max,
+		Reverse: opt.Reverse,
+	}
+}
+
+func (it *IndexIterator) String() string {
+	var min, max, reverse string
+	if !it.Min.Type.IsZero() {
+		min = it.Min.String()
+	}
+	if !it.Max.Type.IsZero() {
+		max = it.Max.String()
+	}
+
+	reverse = "+"
+	if it.Reverse {
+		reverse = "-"
+	}
+
+	return fmt.Sprintf("%s%s[%s:%s]", reverse, it.Name, min, max)
+}
+
+// Iterate over the documents of the table. Each document is stored in the environment
+// that is passed to the fn function, using SetCurrentValue.
+func (it *IndexIterator) Iterate(fn func(env *expr.Environment) error) error {
+	var env expr.Environment
+	env.Params = it.Params
+
+	var min, max []byte
+	var err error
+
+	if !it.Min.Type.IsZero() {
+		min, err = it.Index.EncodeValue(it.Min)
+		if err != nil {
+			return err
+		}
+	}
+
+	if !it.Max.Type.IsZero() {
+		max, err = it.Index.EncodeValue(it.Max)
+		if err != nil {
+			return err
+		}
+	}
+
+	errStop := errors.New("stop")
+
+	if !it.Reverse {
+		if max == nil {
+			return it.Index.AscendGreaterOrEqual(it.Min, func(val, key []byte, isEqual bool) error {
+				d, err := it.Table.GetDocument(key)
+				if err != nil {
+					return err
+				}
+
+				env.SetDocument(d)
+				return fn(&env)
+			})
+		}
+		err := it.Index.AscendGreaterOrEqual(it.Min, func(val, key []byte, isEqual bool) error {
+			// if there is an upper bound, iterate until we reach the max key
+			if bytes.Compare(val, max) >= 0 {
+				return errStop
+			}
+
+			d, err := it.Table.GetDocument(key)
+			if err != nil {
+				return err
+			}
+
+			env.SetDocument(d)
+			return fn(&env)
+		})
+		if err == errStop {
+			err = nil
+		}
+		return err
+	}
+
+	if min == nil {
+		return it.Index.DescendLessOrEqual(it.Max, func(val, key []byte, isEqual bool) error {
+			d, err := it.Table.GetDocument(key)
+			if err != nil {
+				return err
+			}
+
+			env.SetDocument(d)
+			return fn(&env)
+		})
+	}
+
+	err = it.Index.DescendLessOrEqual(it.Max, func(val, key []byte, isEqual bool) error {
+		// if there is a lower bound, iterate until we reach the min key
+		if bytes.Compare(val, min) <= 0 {
+			return errStop
+		}
+
+		d, err := it.Table.GetDocument(key)
+		if err != nil {
+			return err
+		}
+		env.SetDocument(d)
+		return fn(&env)
+	})
+	if err == errStop {
+		err = nil
+	}
+	return err
+}
diff --git a/stream/operator.go b/stream/operator.go
index 24b75a81..4a53528a 100644
--- a/stream/operator.go
+++ b/stream/operator.go
@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/genjidb/genji/database"
 	"github.com/genjidb/genji/document"
 	"github.com/genjidb/genji/sql/query/expr"
 )
@@ -475,3 +476,201 @@ type maxHeap struct {
 func (h maxHeap) Less(i, j int) bool {
 	return bytes.Compare(h.minHeap[i].value, h.minHeap[j].value) > 0
 }
+
+// A TableInsertOperator inserts incoming documents to the table.
+type TableInsertOperator struct {
+	Name  string
+	Table *database.Table
+}
+
+// TableInsert inserts incoming documents to the table.
+func TableInsert(tableName string) *TableInsertOperator {
+	return &TableInsertOperator{Name: tableName}
+}
+
+// Bind the iterator to the table and parameters.
+func (op *TableInsertOperator) Bind(tx *database.Transaction, params []expr.Param) error {
+	var err error
+
+	op.Table, err = tx.GetTable(op.Name)
+	return err
+}
+
+// Op implements the Operator interface.
+func (op *TableInsertOperator) Op() (OperatorFunc, error) {
+	var newEnv expr.Environment
+
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		d, ok := env.GetDocument()
+		if !ok {
+			return nil, errors.New("missing document")
+		}
+
+		d, err := op.Table.Insert(d)
+		if err != nil {
+			return nil, err
+		}
+
+		newEnv.SetDocument(d)
+		newEnv.Outer = env
+		return &newEnv, nil
+	}, nil
+}
+
+func (op *TableInsertOperator) String() string {
+	return fmt.Sprintf("tableInsert('%s')", op.Name)
+}
+
+// A TableReplaceOperator replaces documents in the table
+type TableReplaceOperator struct {
+	Name  string
+	Table *database.Table
+}
+
+// TableReplace replaces documents in the table. Incoming documents must implement the document.Keyer interface.
+func TableReplace(tableName string) *TableReplaceOperator {
+	return &TableReplaceOperator{Name: tableName}
+}
+
+// Bind the iterator to the table and parameters.
+func (op *TableReplaceOperator) Bind(tx *database.Transaction, params []expr.Param) error {
+	var err error
+
+	op.Table, err = tx.GetTable(op.Name)
+	return err
+}
+
+// Op implements the Operator interface.
+func (op *TableReplaceOperator) Op() (OperatorFunc, error) {
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		d, ok := env.GetDocument()
+		if !ok {
+			return nil, errors.New("missing document")
+		}
+
+		ker, ok := d.(document.Keyer)
+		if !ok {
+			return nil, errors.New("missing key")
+		}
+
+		k := ker.RawKey()
+		if k == nil {
+			return nil, errors.New("missing key")
+		}
+
+		err := op.Table.Replace(ker.RawKey(), d)
+		if err != nil {
+			return nil, err
+		}
+
+		return env, nil
+	}, nil
+}
+
+func (op *TableReplaceOperator) String() string {
+	return fmt.Sprintf("tableReplace('%s')", op.Name)
+}
+
+// A TableDeleteOperator replaces documents in the table
+type TableDeleteOperator struct {
+	Name  string
+	Table *database.Table
+}
+
+// TableDelete deletes documents from the table. Incoming documents must implement the document.Keyer interface.
+func TableDelete(tableName string) *TableDeleteOperator {
+	return &TableDeleteOperator{Name: tableName}
+}
+
+// Bind the iterator to the table and parameters.
+func (op *TableDeleteOperator) Bind(tx *database.Transaction, params []expr.Param) error {
+	var err error
+
+	op.Table, err = tx.GetTable(op.Name)
+	return err
+}
+
+// Op implements the Operator interface.
+func (op *TableDeleteOperator) Op() (OperatorFunc, error) {
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		d, ok := env.GetDocument()
+		if !ok {
+			return nil, errors.New("missing document")
+		}
+
+		ker, ok := d.(document.Keyer)
+		if !ok {
+			return nil, errors.New("missing key")
+		}
+
+		k := ker.RawKey()
+		if k == nil {
+			return nil, errors.New("missing key")
+		}
+
+		err := op.Table.Delete(ker.RawKey())
+		if err != nil {
+			return nil, err
+		}
+
+		return env, nil
+	}, nil
+}
+
+func (op *TableDeleteOperator) String() string {
+	return fmt.Sprintf("tableDelete('%s')", op.Name)
+}
+
+// A DistinctOperator filters duplicate documents.
+type DistinctOperator struct{}
+
+// Distinct filters duplicate documents based on one or more expressions.
+func Distinct(exprs ...expr.Expr) *DistinctOperator {
+	return &DistinctOperator{}
+}
+
+// Op implements the Operator interface.
+func (op *DistinctOperator) Op() (OperatorFunc, error) {
+	var buf bytes.Buffer
+	enc := document.NewValueEncoder(&buf)
+	m := make(map[string]struct{})
+
+	return func(env *expr.Environment) (*expr.Environment, error) {
+		buf.Reset()
+
+		d, ok := env.GetDocument()
+		if !ok {
+			return nil, errors.New("missing document")
+		}
+
+		fields, err := document.Fields(d)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, field := range fields {
+			value, err := d.GetByField(field)
+			if err != nil {
+				return nil, err
+			}
+
+			err = enc.Encode(value)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		_, ok = m[string(buf.Bytes())]
+		// if value already exists, filter it out
+		if ok {
+			return nil, nil
+		}
+
+		m[buf.String()] = struct{}{}
+		return env, nil
+	}, nil
+}
+
+func (op *DistinctOperator) String() string {
+	return "distinct()"
+}


Test patch:
diff --git a/database/table_test.go b/database/table_test.go
index c43c97d0..22c9e580 100644
--- a/database/table_test.go
+++ b/database/table_test.go
@@ -118,13 +118,13 @@ func TestTableGetDocument(t *testing.T) {
 		doc1.Add("fieldc", vc)
 		doc2 := newDocument()
 
-		key1, err := tb.Insert(doc1)
+		d, err := tb.Insert(doc1)
 		require.NoError(t, err)
 		_, err = tb.Insert(doc2)
 		require.NoError(t, err)
 
 		// fetch doc1 and make sure it returns the right one
-		res, err := tb.GetDocument(key1)
+		res, err := tb.GetDocument(d.(document.Keyer).RawKey())
 		require.NoError(t, err)
 		fc, err := res.GetByField("fieldc")
 		require.NoError(t, err)
@@ -139,15 +139,15 @@ func TestTableInsert(t *testing.T) {
 		defer cleanup()
 
 		doc := newDocument()
-		key1, err := tb.Insert(doc)
+		d1, err := tb.Insert(doc.Clone())
 		require.NoError(t, err)
-		require.NotEmpty(t, key1)
+		require.NotEmpty(t, d1.(document.Keyer).RawKey())
 
-		key2, err := tb.Insert(doc)
+		d2, err := tb.Insert(doc.Clone())
 		require.NoError(t, err)
-		require.NotEmpty(t, key2)
+		require.NotEmpty(t, d2.(document.Keyer).RawKey())
 
-		require.NotEqual(t, key1, key2)
+		require.NotEqual(t, d1.(document.Keyer).RawKey(), d2.(document.Keyer).RawKey())
 	})
 
 	t.Run("Should generate the right docid on existing databases", func(t *testing.T) {
@@ -168,14 +168,14 @@ func TestTableInsert(t *testing.T) {
 			require.NoError(t, err)
 
 			doc := newDocument()
-			key, err := tb.Insert(doc)
+			d, err := tb.Insert(doc)
 			require.NoError(t, err)
-			require.NotEmpty(t, key)
+			require.NotEmpty(t, d.(document.Keyer).RawKey())
 
 			err = tx.Commit()
 			require.NoError(t, err)
 
-			return key
+			return d.(document.Keyer).RawKey()
 		}
 
 		key1 := insertDoc(db)
@@ -216,16 +216,16 @@ func TestTableInsert(t *testing.T) {
 		require.NoError(t, err)
 
 		// insert
-		k, err := tb.Insert(doc)
+		d, err := tb.Insert(doc)
 		require.NoError(t, err)
-		require.Equal(t, binarysort.AppendInt64(nil, 10), k)
+		require.Equal(t, binarysort.AppendInt64(nil, 10), d.(document.Keyer).RawKey())
 
 		// make sure the document is fetchable using the returned key
-		_, err = tb.GetDocument(k)
+		_, err = tb.GetDocument(d.(document.Keyer).RawKey())
 		require.NoError(t, err)
 
 		// insert again
-		k, err = tb.Insert(doc)
+		d, err = tb.Insert(doc)
 		require.Equal(t, database.ErrDuplicateDocument, err)
 	})
 
@@ -248,10 +248,10 @@ func TestTableInsert(t *testing.T) {
 		require.NoError(t, err)
 
 		// insert
-		key, err := tb.Insert(doc)
+		d, err := tb.Insert(doc)
 		require.NoError(t, err)
 
-		d, err := tb.GetDocument(key)
+		d, err = tb.GetDocument(d.(document.Keyer).RawKey())
 		require.NoError(t, err)
 
 		v, err := parsePath(t, "foo[0]").GetValueFromDocument(d)
@@ -314,9 +314,9 @@ func TestTableInsert(t *testing.T) {
 		// create one document without the foo field
 		doc2 := newDocument()
 
-		key1, err := tb.Insert(doc1)
+		d1, err := tb.Insert(doc1)
 		require.NoError(t, err)
-		key2, err := tb.Insert(doc2)
+		d2, err := tb.Insert(doc2)
 		require.NoError(t, err)
 
 		var count int
@@ -325,9 +325,9 @@ func TestTableInsert(t *testing.T) {
 			case 0:
 				// key2, which doesn't countain the field must appear first in the next,
 				// as null values are the smallest possible values
-				require.Equal(t, key2, k)
+				require.Equal(t, d2.(document.Keyer).RawKey(), k)
 			case 1:
-				require.Equal(t, key1, k)
+				require.Equal(t, d1.(document.Keyer).RawKey(), k)
 			}
 			count++
 			return nil
@@ -357,11 +357,11 @@ func TestTableInsert(t *testing.T) {
 			Add("bat", document.NewIntegerValue(20))
 
 		// insert
-		key, err := tb.Insert(doc)
+		d, err := tb.Insert(doc)
 		require.NoError(t, err)
 
 		// make sure the fields have been converted to the right types
-		d, err := tb.GetDocument(key)
+		d, err = tb.GetDocument(d.(document.Keyer).RawKey())
 		require.NoError(t, err)
 		v, err := d.GetByField("foo")
 		require.NoError(t, err)
@@ -478,11 +478,11 @@ func TestTableInsert(t *testing.T) {
 		require.NoError(t, err)
 
 		// insert with empty foo field shouldn't fail
-		k, err := tb1.Insert(document.NewFieldBuffer().
+		d, err := tb1.Insert(document.NewFieldBuffer().
 			Add("bar", document.NewDoubleValue(1)))
 		require.NoError(t, err)
 
-		d, err := tb1.GetDocument(k)
+		d, err = tb1.GetDocument(d.(document.Keyer).RawKey())
 		require.NoError(t, err)
 		v, err := d.GetByField("foo")
 		require.NoError(t, err)
@@ -499,11 +499,11 @@ func TestTableInsert(t *testing.T) {
 		require.NoError(t, err)
 
 		// insert with empty foo field shouldn't fail
-		k, err = tb2.Insert(document.NewFieldBuffer().
+		d, err = tb2.Insert(document.NewFieldBuffer().
 			Add("bar", document.NewIntegerValue(1)))
 		require.NoError(t, err)
 
-		d, err = tb2.GetDocument(k)
+		d, err = tb2.GetDocument(d.(document.Keyer).RawKey())
 		require.NoError(t, err)
 		v, err = d.GetByField("foo")
 		require.NoError(t, err)
@@ -563,21 +563,21 @@ func TestTableDelete(t *testing.T) {
 		doc1.Add("fieldc", document.NewIntegerValue(40))
 		doc2 := newDocument()
 
-		key1, err := tb.Insert(doc1)
+		d1, err := tb.Insert(doc1.Clone())
 		require.NoError(t, err)
-		key2, err := tb.Insert(doc2)
+		d2, err := tb.Insert(doc2.Clone())
 		require.NoError(t, err)
 
 		// delete the document
-		err = tb.Delete([]byte(key1))
+		err = tb.Delete([]byte(d1.(document.Keyer).RawKey()))
 		require.NoError(t, err)
 
 		// try again, should fail
-		err = tb.Delete([]byte(key1))
+		err = tb.Delete([]byte(d1.(document.Keyer).RawKey()))
 		require.Equal(t, database.ErrDocumentNotFound, err)
 
 		// make sure it didn't also delete the other one
-		res, err := tb.GetDocument(key2)
+		res, err := tb.GetDocument(d2.(document.Keyer).RawKey())
 		require.NoError(t, err)
 		_, err = res.GetByField("fieldc")
 		require.Error(t, err)
@@ -604,9 +604,9 @@ func TestTableReplace(t *testing.T) {
 			Add("fielda", document.NewTextValue("c")).
 			Add("fieldb", document.NewTextValue("d"))
 
-		key1, err := tb.Insert(doc1)
+		d1, err := tb.Insert(doc1)
 		require.NoError(t, err)
-		key2, err := tb.Insert(doc2)
+		d2, err := tb.Insert(doc2)
 		require.NoError(t, err)
 
 		// create a third document
@@ -615,18 +615,18 @@ func TestTableReplace(t *testing.T) {
 			Add("fieldb", document.NewTextValue("f"))
 
 		// replace doc1 with doc3
-		err = tb.Replace(key1, doc3)
+		err = tb.Replace(d1.(document.Keyer).RawKey(), doc3)
 		require.NoError(t, err)
 
 		// make sure it replaced it cordoctly
-		res, err := tb.GetDocument(key1)
+		res, err := tb.GetDocument(d1.(document.Keyer).RawKey())
 		require.NoError(t, err)
 		f, err := res.GetByField("fielda")
 		require.NoError(t, err)
 		require.Equal(t, "e", f.V.(string))
 
 		// make sure it didn't also replace the other one
-		res, err = tb.GetDocument(key2)
+		res, err = tb.GetDocument(d2.(document.Keyer).RawKey())
 		require.NoError(t, err)
 		f, err = res.GetByField("fielda")
 		require.NoError(t, err)
diff --git a/stream/iterator_test.go b/stream/iterator_test.go
new file mode 100644
index 00000000..44f533f6
--- /dev/null
+++ b/stream/iterator_test.go
@@ -0,0 +1,294 @@
+package stream_test
+
+import (
+	"testing"
+
+	"github.com/genjidb/genji"
+	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/sql/query/expr"
+	"github.com/genjidb/genji/stream"
+	"github.com/genjidb/genji/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTableIterator(t *testing.T) {
+	tests := []struct {
+		name                  string
+		docsInTable, expected testutil.Docs
+		min, max              *document.Value
+		reverse               bool
+		fails                 bool
+	}{
+		{name: "empty"},
+		{
+			"no range",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			nil, nil, false, false,
+		},
+		{
+			"max",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 1}`),
+			nil, testutil.MakeValue(t, 2),
+			false, false,
+		},
+		{
+			"min",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeValue(t, 1), nil,
+			false, false,
+		},
+		{
+			"min/max",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 1}`),
+			testutil.MakeValue(t, 1), testutil.MakeValue(t, 2),
+			false, false,
+		},
+		{
+			"reverse/no range",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 2}`, `{"a": 1}`),
+			nil, nil, true, false,
+		},
+		{
+			"reverse/max",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 2}`, `{"a": 1}`),
+			nil, testutil.MakeValue(t, 2),
+			true, false,
+		},
+		{
+			"reverse/min",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 2}`),
+			testutil.MakeValue(t, 1), nil,
+			true, false,
+		},
+		{
+			"reverse/min/max",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 2}`),
+			testutil.MakeValue(t, 1), testutil.MakeValue(t, 2),
+			true, false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			db, err := genji.Open(":memory:")
+			require.NoError(t, err)
+			defer db.Close()
+
+			err = db.Exec("CREATE TABLE test (a INTEGER)")
+			require.NoError(t, err)
+
+			for _, doc := range test.docsInTable {
+				err = db.Exec("INSERT INTO test VALUES ?", doc)
+				require.NoError(t, err)
+			}
+
+			tx, err := db.Begin(false)
+			require.NoError(t, err)
+			defer tx.Rollback()
+
+			opts := stream.TableIteratorOptions{
+				Reverse: test.reverse,
+			}
+			if test.min != nil {
+				opts.Min = *test.min
+			}
+			if test.max != nil {
+				opts.Max = *test.max
+			}
+			it := stream.NewTableIteratorWithOptions("test", opts)
+			err = it.Bind(tx.Transaction, []expr.Param{{Name: "foo", Value: 1}})
+			require.NoError(t, err)
+			s := stream.New(it)
+
+			var i int
+			var got testutil.Docs
+			err = s.Iterate(func(env *expr.Environment) error {
+				d, ok := env.GetDocument()
+				require.True(t, ok)
+				var fb document.FieldBuffer
+
+				err = fb.Copy(d)
+				require.NoError(t, err)
+
+				got = append(got, &fb)
+				v, err := env.GetParamByName("foo")
+				require.NoError(t, err)
+				require.Equal(t, document.NewIntegerValue(1), v)
+				i++
+				return nil
+			})
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, len(test.expected), i)
+				test.expected.RequireEqual(t, got)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, `+test[1:2]`, stream.NewTableIteratorWithOptions("test", stream.TableIteratorOptions{
+			Min: *testutil.MakeValue(t, 1),
+			Max: *testutil.MakeValue(t, 2),
+		}).String())
+
+		require.Equal(t, `-test[1:]`, stream.NewTableIteratorWithOptions("test", stream.TableIteratorOptions{
+			Min:     *testutil.MakeValue(t, 1),
+			Reverse: true,
+		}).String())
+
+		require.Equal(t, `-test[:]`, stream.NewTableIteratorWithOptions("test", stream.TableIteratorOptions{
+			Reverse: true,
+		}).String())
+	})
+}
+
+func TestIndexIterator(t *testing.T) {
+	tests := []struct {
+		name                  string
+		docsInTable, expected testutil.Docs
+		min, max              *document.Value
+		reverse               bool
+		fails                 bool
+	}{
+		{name: "empty"},
+		{
+			"no range",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			nil, nil, false, false,
+		},
+		{
+			"max",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 1}`),
+			nil, testutil.MakeValue(t, 2),
+			false, false,
+		},
+		{
+			"min",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeValue(t, 1), nil,
+			false, false,
+		},
+		{
+			"min/max",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 1}`),
+			testutil.MakeValue(t, 1), testutil.MakeValue(t, 2),
+			false, false,
+		},
+		{
+			"reverse/no range",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 2}`, `{"a": 1}`),
+			nil, nil, true, false,
+		},
+		{
+			"reverse/max",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 2}`, `{"a": 1}`),
+			nil, testutil.MakeValue(t, 2),
+			true, false,
+		},
+		{
+			"reverse/min",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 2}`),
+			testutil.MakeValue(t, 1), nil,
+			true, false,
+		},
+		{
+			"reverse/min/max",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`),
+			testutil.MakeDocuments(`{"a": 2}`),
+			testutil.MakeValue(t, 1), testutil.MakeValue(t, 2),
+			true, false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			db, err := genji.Open(":memory:")
+			require.NoError(t, err)
+			defer db.Close()
+
+			err = db.Exec("CREATE TABLE test (a INTEGER); CREATE INDEX idx_test_a ON test(a)")
+			require.NoError(t, err)
+
+			for _, doc := range test.docsInTable {
+				err = db.Exec("INSERT INTO test VALUES ?", doc)
+				require.NoError(t, err)
+			}
+
+			tx, err := db.Begin(false)
+			require.NoError(t, err)
+			defer tx.Rollback()
+
+			opts := stream.IndexIteratorOptions{
+				Reverse: test.reverse,
+			}
+			if test.min != nil {
+				opts.Min = *test.min
+			}
+			if test.max != nil {
+				opts.Max = *test.max
+			}
+			it := stream.NewIndexIteratorWithOptions("idx_test_a", opts)
+			err = it.Bind(tx.Transaction, []expr.Param{{Name: "foo", Value: 1}})
+			require.NoError(t, err)
+			s := stream.New(it)
+
+			var i int
+			var got testutil.Docs
+			err = s.Iterate(func(env *expr.Environment) error {
+				d, ok := env.GetDocument()
+				require.True(t, ok)
+				var fb document.FieldBuffer
+
+				err = fb.Copy(d)
+				require.NoError(t, err)
+
+				got = append(got, &fb)
+				v, err := env.GetParamByName("foo")
+				require.NoError(t, err)
+				require.Equal(t, document.NewIntegerValue(1), v)
+				i++
+				return nil
+			})
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, len(test.expected), i)
+				test.expected.RequireEqual(t, got)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, `+test[1:2]`, stream.NewIndexIteratorWithOptions("test", stream.IndexIteratorOptions{
+			Min: *testutil.MakeValue(t, 1),
+			Max: *testutil.MakeValue(t, 2),
+		}).String())
+
+		require.Equal(t, `-test[1:]`, stream.NewIndexIteratorWithOptions("test", stream.IndexIteratorOptions{
+			Min:     *testutil.MakeValue(t, 1),
+			Reverse: true,
+		}).String())
+
+		require.Equal(t, `-test[:]`, stream.NewIndexIteratorWithOptions("test", stream.IndexIteratorOptions{
+			Reverse: true,
+		}).String())
+	})
+}
diff --git a/stream/operator_test.go b/stream/operator_test.go
index 775568fa..27fd0f6e 100644
--- a/stream/operator_test.go
+++ b/stream/operator_test.go
@@ -5,21 +5,22 @@ import (
 	"strconv"
 	"testing"
 
+	"github.com/genjidb/genji"
 	"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/genjidb/genji/testutil"
 	"github.com/stretchr/testify/require"
 )
 
-func docFromJSON(d string) document.Document {
+func docFromJSON(d string) *document.FieldBuffer {
 	var fb document.FieldBuffer
 
 	err := fb.UnmarshalJSON([]byte(d))
 	if err != nil {
 		panic(err)
 	}
-
 	return &fb
 }
 
@@ -330,7 +331,9 @@ func TestReduce(t *testing.T) {
 			err := s.Iterate(func(env *expr.Environment) error {
 				d, ok := env.GetDocument()
 				require.True(t, ok)
-				got = append(got, d)
+				var fb document.FieldBuffer
+				fb.Copy(d)
+				got = append(got, &fb)
 				return nil
 			})
 			if test.fails {
@@ -419,3 +422,268 @@ func TestSort(t *testing.T) {
 		require.Equal(t, `sort(a)`, stream.Sort(parser.MustParseExpr("a")).String())
 	})
 }
+
+func TestTableInsert(t *testing.T) {
+	tests := []struct {
+		name    string
+		in, out *expr.Environment
+		docid   int
+		fails   bool
+	}{
+		{
+			"doc with no key",
+			expr.NewEnvironment(docFromJSON(`{"a": 10}`)),
+			expr.NewEnvironment(docFromJSON(`{"a": 10}`)),
+			1,
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			db, err := genji.Open(":memory:")
+			require.NoError(t, err)
+			defer db.Close()
+
+			err = db.Exec("CREATE TABLE test (a INTEGER)")
+			require.NoError(t, err)
+
+			tx, err := db.Begin(true)
+			require.NoError(t, err)
+			defer tx.Rollback()
+
+			if test.out != nil {
+				test.out.Outer = test.in
+				tb, err := tx.GetTable("test")
+				require.NoError(t, err)
+				k, err := tb.EncodeValueToKey(document.NewIntegerValue(1))
+				require.NoError(t, err)
+				test.out.Doc.(*document.FieldBuffer).EncodedKey = k
+			}
+
+			ti := stream.TableInsert("test")
+			err = ti.Bind(tx.Transaction, nil)
+			require.NoError(t, err)
+
+			op, err := ti.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.TableInsert("test").String(), "tableInsert('test')")
+	})
+}
+
+func TestTableReplace(t *testing.T) {
+	tests := []struct {
+		name                  string
+		docsInTable, expected testutil.Docs
+		in                    *expr.Environment
+		fails                 bool
+	}{
+		{
+			"doc with key",
+			testutil.MakeDocuments(`{"a": 1, "b": 1}`),
+			testutil.MakeDocuments(`{"a": 1, "b": 2}`),
+			expr.NewEnvironment(docFromJSON(`{"a": 1, "b": 2}`)),
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			db, err := genji.Open(":memory:")
+			require.NoError(t, err)
+			defer db.Close()
+
+			err = db.Exec("CREATE TABLE test (a INTEGER PRIMARY KEY, b INTEGER)")
+			require.NoError(t, err)
+
+			for _, doc := range test.docsInTable {
+				err = db.Exec("INSERT INTO test VALUES ?", doc)
+				require.NoError(t, err)
+			}
+
+			tx, err := db.Begin(true)
+			require.NoError(t, err)
+			defer tx.Rollback()
+
+			tb, err := tx.GetTable("test")
+			require.NoError(t, err)
+			kk, err := test.in.Doc.GetByField("a")
+			require.NoError(t, err)
+
+			k, err := tb.EncodeValueToKey(kk)
+			require.NoError(t, err)
+			test.in.Doc.(*document.FieldBuffer).EncodedKey = k
+
+			ti := stream.TableReplace("test")
+			err = ti.Bind(tx.Transaction, nil)
+			require.NoError(t, err)
+
+			op, err := ti.Op()
+			require.NoError(t, err)
+			env, err := op(test.in)
+			if test.fails {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, test.in, env)
+
+			res, err := tx.Query("SELECT * FROM test")
+			require.NoError(t, err)
+			defer res.Close()
+
+			var got []document.Document
+			err = res.Iterate(func(d document.Document) error {
+				var fb document.FieldBuffer
+				fb.Copy(d)
+				got = append(got, fb)
+				return nil
+			})
+			require.NoError(t, err)
+			test.expected.RequireEqual(t, got)
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, stream.TableReplace("test").String(), "tableReplace('test')")
+	})
+}
+
+func TestTableDelete(t *testing.T) {
+	tests := []struct {
+		name                  string
+		docsInTable, expected testutil.Docs
+		in                    *expr.Environment
+		fails                 bool
+	}{
+		{
+			"doc with key",
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 2}`, `{"a": 3}`),
+			testutil.MakeDocuments(`{"a": 1}`, `{"a": 3}`),
+			expr.NewEnvironment(docFromJSON(`{"a": 2}`)),
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			db, err := genji.Open(":memory:")
+			require.NoError(t, err)
+			defer db.Close()
+
+			err = db.Exec("CREATE TABLE test (a INTEGER PRIMARY KEY)")
+			require.NoError(t, err)
+
+			for _, doc := range test.docsInTable {
+				err = db.Exec("INSERT INTO test VALUES ?", doc)
+				require.NoError(t, err)
+			}
+
+			tx, err := db.Begin(true)
+			require.NoError(t, err)
+			defer tx.Rollback()
+
+			tb, err := tx.GetTable("test")
+			require.NoError(t, err)
+			kk, err := test.in.Doc.GetByField("a")
+			require.NoError(t, err)
+
+			k, err := tb.EncodeValueToKey(kk)
+			require.NoError(t, err)
+			test.in.Doc.(*document.FieldBuffer).EncodedKey = k
+
+			ti := stream.TableDelete("test")
+			err = ti.Bind(tx.Transaction, nil)
+			require.NoError(t, err)
+
+			op, err := ti.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.in, env)
+			}
+
+			res, err := tx.Query("SELECT * FROM test")
+			require.NoError(t, err)
+			defer res.Close()
+
+			var got []document.Document
+			err = res.Iterate(func(d document.Document) error {
+				var fb document.FieldBuffer
+				fb.Copy(d)
+				got = append(got, fb)
+				return nil
+			})
+			require.NoError(t, err)
+			test.expected.RequireEqual(t, got)
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, stream.TableDelete("test").String(), "tableDelete('test')")
+	})
+}
+
+func TestDistinct(t *testing.T) {
+	tests := []struct {
+		name   string
+		values testutil.Docs
+		want   testutil.Docs
+		fails  bool
+	}{
+		{
+			"all different",
+			testutil.MakeDocuments(`{"a": 0}`, `{"a": null}`, `{"a": true}`),
+			testutil.MakeDocuments(`{"a": 0}`, `{"a": null}`, `{"a": true}`),
+			false,
+		},
+		{
+			"some duplicates",
+			testutil.MakeDocuments(`{"a": 0}`, `{"a": 0}`, `{"a": null}`, `{"a": null}`, `{"a": true}`, `{"a": true}`, `{"a": [1, 2]}`, `{"a": [1, 2]}`),
+			testutil.MakeDocuments(`{"a": 0}`, `{"a": null}`, `{"a": true}`, `{"a": [1, 2]}`),
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s := stream.New(stream.NewDocumentIterator(test.values...))
+			s = s.Pipe(stream.Distinct())
+
+			var got []document.Document
+			err := s.Iterate(func(env *expr.Environment) error {
+				d, ok := env.GetDocument()
+				require.True(t, ok)
+				var fb document.FieldBuffer
+				err := fb.Copy(d)
+				require.NoError(t, err)
+				got = append(got, &fb)
+				return nil
+			})
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				test.want.RequireEqual(t, got)
+			}
+		})
+	}
+
+	t.Run("String", func(t *testing.T) {
+		require.Equal(t, `distinct()`, stream.Distinct().String())
+	})
+}
diff --git a/testutil/document.go b/testutil/document.go
new file mode 100644
index 00000000..6d71d0b3
--- /dev/null
+++ b/testutil/document.go
@@ -0,0 +1,49 @@
+package testutil
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/genjidb/genji/document"
+	"github.com/stretchr/testify/require"
+)
+
+// MakeValue turns v into a document.Value.
+func MakeValue(t testing.TB, v interface{}) *document.Value {
+	t.Helper()
+
+	vv, err := document.NewValue(v)
+	require.NoError(t, err)
+	return &vv
+}
+
+// MakeDocument creates a document from a json string.
+func MakeDocument(jsonDoc string) document.Document {
+	return document.NewFromJSON([]byte(jsonDoc))
+}
+
+// MakeDocuments creates a slice of document from json strings.
+func MakeDocuments(jsonDocs ...string) (docs Docs) {
+	for _, jsonDoc := range jsonDocs {
+		docs = append(docs, MakeDocument(jsonDoc))
+	}
+	return
+}
+
+type Docs []document.Document
+
+func (docs Docs) RequireEqual(t testing.TB, others Docs) {
+	t.Helper()
+
+	require.Equal(t, len(docs), len(others), fmt.Sprintf("expected len %d, got %d", len(docs), len(others)))
+
+	for i, d := range docs {
+		l := document.NewDocumentValue(d)
+		r := document.NewDocumentValue(others[i])
+		ok, err := l.IsEqual(r)
+		require.NoError(t, err)
+		if !ok {
+			require.Equal(t, l, r)
+		}
+	}
+}