From d16978f5dcc0da17b1607f9c284ae18d89d4294c Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Fri, 30 Jul 2021 07:14:15 -0600
Subject: [PATCH 1/2] Add max memory limit to Extract() to prevent OOM

This commit changes the Extract() query to return an error if the
result set gets too large in order to prevent out-of-memory (OOM)
panics.
---
 api.go                  |   1 +
 encoding/proto/proto.go |   2 +
 executor.go             | 160 +++++++++++++++++++--------
 executor_test.go        |  70 ++++++++++++
 handler.go              |   3 +
 pb/public.pb.go         | 233 +++++++++++++++++++++++-----------------
 pb/public.proto         |   1 +
 server.go               |  10 +-
 8 files changed, 338 insertions(+), 142 deletions(-)

diff --git a/api.go b/api.go
index c433503db..7c3b17250 100644
--- a/api.go
+++ b/api.go
@@ -197,6 +197,7 @@ func (api *API) query(ctx context.Context, req *QueryRequest) (QueryResponse, er
 		Profile:       req.Profile,
 		PreTranslated: req.PreTranslated,
 		EmbeddedData:  req.EmbeddedData, // precomputed values that needed to be passed with the request
+		MaxMemory:     req.MaxMemory,
 	}
 	resp, err := api.server.executor.Execute(ctx, req.Index, q, req.Shards, execOpts)
 	if err != nil {
diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go
index 5904f82e8..69a456904 100644
--- a/encoding/proto/proto.go
+++ b/encoding/proto/proto.go
@@ -484,6 +484,7 @@ func (s Serializer) encodeQueryRequest(m *pilosa.QueryRequest) *pb.QueryRequest
 		Remote:        m.Remote,
 		PreTranslated: m.PreTranslated,
 		EmbeddedData:  make([]*pb.Row, len(m.EmbeddedData)),
+		MaxMemory:     m.MaxMemory,
 	}
 	for i := range m.EmbeddedData {
 		r.EmbeddedData[i] = s.encodeRow(m.EmbeddedData[i])
@@ -1186,6 +1187,7 @@ func (s Serializer) decodeQueryRequest(pb *pb.QueryRequest, m *pilosa.QueryReque
 	m.Remote = pb.Remote
 	m.EmbeddedData = make([]*pilosa.Row, len(pb.EmbeddedData))
 	m.PreTranslated = pb.PreTranslated
+	m.MaxMemory = pb.MaxMemory
 	for i := range pb.EmbeddedData {
 		m.EmbeddedData[i] = s.decodeRow(pb.EmbeddedData[i])
 	}
diff --git a/executor.go b/executor.go
index 9283bb2f6..b0d1959d4 100644
--- a/executor.go
+++ b/executor.go
@@ -25,11 +25,10 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 	"unsafe"
 
-	"golang.org/x/sync/errgroup"
-
 	"github.com/lib/pq"
 	"github.com/molecula/featurebase/v2/disco"
 	"github.com/molecula/featurebase/v2/pql"
@@ -40,6 +39,7 @@ import (
 	"github.com/molecula/featurebase/v2/topology"
 	"github.com/molecula/featurebase/v2/tracing"
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 )
 
 // defaultField is the field used if one is not specified.
@@ -75,6 +75,9 @@ type executor struct {
 	workersWG      sync.WaitGroup
 	workerPoolSize int
 	work           chan job
+
+	// Maximum per-request memory usage (Extract() only)
+	maxMemory int64
 }
 
 // executorOption is a functional option type for pilosa.Executor
@@ -94,6 +97,13 @@ func optExecutorWorkerPoolSize(size int) executorOption {
 	}
 }
 
+func optExecutorMaxMemory(v int64) executorOption {
+	return func(e *executor) error {
+		e.maxMemory = v
+		return nil
+	}
+}
+
 func emptyResult(c *pql.Call) interface{} {
 	switch c.Name {
 	case "Clear", "ClearRow":
@@ -839,7 +849,7 @@ func (e *executor) executeIncludesColumnCall(ctx context.Context, qcx *Qcx, inde
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeIncludesColumnCallShard(ctx, qcx, index, c, shard, col)
 	}
 
@@ -888,7 +898,7 @@ func (e *executor) executeFieldValueCall(ctx context.Context, qcx *Qcx, index st
 	shard := colID / ShardWidth
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeFieldValueCallShard(ctx, qcx, field, colID, shard)
 	}
 
@@ -1053,7 +1063,7 @@ func (e *executor) executeSum(ctx context.Context, qcx *Qcx, index string, c *pq
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeSumCountShard(ctx, qcx, index, c, nil, shard)
 	}
 
@@ -1106,7 +1116,7 @@ func (e *executor) executeDistinct(ctx context.Context, qcx *Qcx, index string,
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeDistinctShard(ctx, qcx, index, field, c, shard)
 	}
 
@@ -1157,7 +1167,7 @@ func (e *executor) executeMin(ctx context.Context, qcx *Qcx, index string, c *pq
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeMinShard(ctx, qcx, index, c, shard)
 	}
 
@@ -1193,7 +1203,7 @@ func (e *executor) executeMax(ctx context.Context, qcx *Qcx, index string, c *pq
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeMaxShard(ctx, qcx, index, c, shard)
 	}
 
@@ -1348,7 +1358,7 @@ func (e *executor) executeMinRow(ctx context.Context, qcx *Qcx, index string, c
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeMinRowShard(ctx, qcx, index, c, shard)
 	}
 
@@ -1387,7 +1397,7 @@ func (e *executor) executeMaxRow(ctx context.Context, qcx *Qcx, index string, c
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeMaxRowShard(ctx, qcx, index, c, shard)
 	}
 
@@ -1445,7 +1455,7 @@ func (e *executor) executeBitmapCall(ctx context.Context, qcx *Qcx, index string
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeBitmapCallShard(ctx, qcx, index, c, shard)
 	}
 
@@ -1977,7 +1987,7 @@ func (e *executor) executeTopK(ctx context.Context, qcx *Qcx, index string, c *p
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeTopK")
 	defer span.Finish()
 
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeTopKShard(ctx, qcx, index, c, shard)
 	}
 
@@ -2454,7 +2464,7 @@ func (e *executor) executeTopNShards(ctx context.Context, qcx *Qcx, index string
 	defer span.Finish()
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeTopNShard(ctx, qcx, index, c, shard)
 	}
 
@@ -2906,7 +2916,7 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
 
 	ignoreLimit := sorter != nil
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeGroupByShard(ctx, qcx, index, c, filter, shard, childRows, bases, ignoreLimit)
 	}
 
@@ -3541,7 +3551,7 @@ func (e *executor) executeRows(ctx context.Context, qcx *Qcx, index string, c *p
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeRowsShard(ctx, qcx, index, fieldName, c, shard)
 	}
 
@@ -4153,6 +4163,11 @@ func (e *executor) executeExternalLookup(ctx context.Context, qcx *Qcx, index st
 }
 
 func (e *executor) executeExtract(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (ExtractedIDMatrix, error) {
+	// Defualt maximum memory, if not passed in.
+	if opt.MaxMemory == 0 {
+		opt.MaxMemory = e.maxMemory
+	}
+
 	// Extract the column filter call.
 	if len(c.Children) < 1 {
 		return ExtractedIDMatrix{}, errors.New("missing column filter in Extract")
@@ -4183,8 +4198,8 @@ func (e *executor) executeExtract(ctx context.Context, qcx *Qcx, index string, c
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
-		return e.executeExtractShard(ctx, qcx, index, fields, filter, shard)
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
+		return e.executeExtractShard(ctx, qcx, index, fields, filter, shard, mopt)
 	}
 
 	// Merge returned results at coordinating node.
@@ -4218,7 +4233,7 @@ func mergeBits(bits *Row, mask uint64, out map[uint64]uint64) {
 var trueRowFakeID = []uint64{1}
 var falseRowFakeID = []uint64{0}
 
-func (e *executor) executeExtractShard(ctx context.Context, qcx *Qcx, index string, fields []string, filter *pql.Call, shard uint64) (_ ExtractedIDMatrix, err0 error) {
+func (e *executor) executeExtractShard(ctx context.Context, qcx *Qcx, index string, fields []string, filter *pql.Call, shard uint64, mopt *mapOptions) (_ ExtractedIDMatrix, err0 error) {
 
 	// Execute filter.
 	colsBitmap, err := e.executeBitmapCallShard(ctx, qcx, index, filter, shard)
@@ -4399,10 +4414,14 @@ func (e *executor) executeExtractShard(ctx context.Context, qcx *Qcx, index stri
 
 	// Emit the final matrix.
 	// Like RowIDs, this is an internal type and will need to be converted.
-	return ExtractedIDMatrix{
+	matrix := ExtractedIDMatrix{
 		Fields:  fields,
 		Columns: m,
-	}, nil
+	}
+	if v := atomic.AddInt64(mopt.memoryAvailable, -calcResultMemory(matrix)); v < 0 {
+		return ExtractedIDMatrix{}, fmt.Errorf("result exceeds available memory")
+	}
+	return matrix, nil
 }
 
 func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ *Row, err0 error) {
@@ -4993,7 +5012,7 @@ func (e *executor) executeCount(ctx context.Context, qcx *Qcx, index string, c *
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		row, err := e.executeBitmapCallShard(ctx, qcx, index, child, shard)
 		if err != nil {
 			return 0, err
@@ -5097,7 +5116,7 @@ func (e *executor) executeClearBitField(ctx context.Context, qcx *Qcx, index str
 		}
 
 		// Forward call to remote node otherwise.
-		res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil)
+		res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil, 0)
 		if err != nil {
 			return false, err
 		}
@@ -5130,7 +5149,7 @@ func (e *executor) executeClearRow(ctx context.Context, qcx *Qcx, index string,
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeClearRowShard(ctx, qcx, index, c, shard)
 	}
 
@@ -5225,7 +5244,7 @@ func (e *executor) executeSetRow(ctx context.Context, qcx *Qcx, indexName string
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeSetRowShard(ctx, qcx, indexName, c, shard)
 	}
 
@@ -5457,7 +5476,7 @@ func (e *executor) executeSetBitField(ctx context.Context, qcx *Qcx, index strin
 		}
 
 		// Forward call to remote node otherwise.
-		res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil)
+		res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil, 0)
 		if err != nil {
 			return false, err
 		}
@@ -5504,7 +5523,7 @@ func (e *executor) executeSetValueField(ctx context.Context, qcx *Qcx, index str
 		}
 
 		// Forward call to remote node otherwise.
-		res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil)
+		res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil, 0)
 		if err != nil {
 			return false, err
 		}
@@ -5549,7 +5568,7 @@ func (e *executor) executeClearValueField(ctx context.Context, qcx *Qcx, index s
 		}
 
 		// Forward call to remote node otherwise.
-		res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil)
+		res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil, 0)
 		if err != nil {
 			return false, err
 		}
@@ -5559,7 +5578,7 @@ func (e *executor) executeClearValueField(ctx context.Context, qcx *Qcx, index s
 }
 
 // remoteExec executes a PQL query remotely for a set of shards on a node.
-func (e *executor) remoteExec(ctx context.Context, node *topology.Node, index string, q *pql.Query, shards []uint64, embed []*Row) (results []interface{}, err error) { // nolint: interfacer
+func (e *executor) remoteExec(ctx context.Context, node *topology.Node, index string, q *pql.Query, shards []uint64, embed []*Row, maxMemory int64) (results []interface{}, err error) { // nolint: interfacer
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeExec")
 	defer span.Finish()
 
@@ -5569,6 +5588,7 @@ func (e *executor) remoteExec(ctx context.Context, node *topology.Node, index st
 		Shards:       shards,
 		Remote:       true,
 		EmbeddedData: embed,
+		MaxMemory:    maxMemory,
 	}
 
 	resp, err := e.client.QueryNode(ctx, &node.URI, index, pbreq)
@@ -5744,7 +5764,7 @@ func makeEmbeddedDataForShards(allRows []*Row, shards []uint64) []*Row {
 	return newRows
 }
 
-func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapResponse, nodes []*topology.Node, index string, shards []uint64, c *pql.Call, opt *execOptions, lastAttempt bool, mapFn mapFunc, reduceFn reduceFunc) error {
+func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapResponse, nodes []*topology.Node, index string, shards []uint64, c *pql.Call, opt *execOptions, lastAttempt bool, mapFn mapFunc, reduceFn reduceFunc) (reterr error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapper")
 	defer span.Finish()
 
@@ -5756,26 +5776,47 @@ func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapRe
 	done := ctx.Done()
 
 	// Execute each node in a separate goroutine.
+	var memoryUsed int64
+	var mu sync.Mutex
 	for n, nodeShards := range m {
 		n := n
 		nodeShards := nodeShards
 		eg.Go(func() error {
+			// Execute serially max memory is specified.
+			if opt.MaxMemory > 0 {
+				mu.Lock()
+				defer mu.Unlock()
+			}
+
 			resp := mapResponse{node: n, shards: nodeShards}
 
+			// Calculate remaining memory. This applies to Extract() only.
+			// Default to a high number if we are not tracking memory.
+			memoryAvailable := opt.MaxMemory - atomic.LoadInt64(&memoryUsed)
+			if opt.MaxMemory <= 0 {
+				memoryAvailable = math.MaxInt64
+			}
+
 			// Send local shards to mapper, otherwise remote exec.
 			if n.ID == e.Node.ID {
-				resp.result, resp.err = e.mapperLocal(ctx, nodeShards, mapFn, reduceFn)
+				resp.result, resp.err = e.mapperLocal(ctx, nodeShards, mapFn, reduceFn, memoryAvailable)
 			} else if !opt.Remote {
 				var embeddedRowsForNode []*Row
 				if opt.EmbeddedData != nil {
 					embeddedRowsForNode = makeEmbeddedDataForShards(opt.EmbeddedData, nodeShards)
 				}
-				results, err := e.remoteExec(ctx, n, index, &pql.Query{Calls: []*pql.Call{c}}, nodeShards, embeddedRowsForNode)
+				results, err := e.remoteExec(ctx, n, index, &pql.Query{Calls: []*pql.Call{c}}, nodeShards, embeddedRowsForNode, memoryAvailable)
 				if len(results) > 0 {
 					resp.result = results[0]
 				}
 				resp.err = err
 			}
+
+			// Track total memory used in response.
+			if v := atomic.AddInt64(&memoryUsed, calcResultMemory(resp.result)); opt.MaxMemory > 0 && v > opt.MaxMemory {
+				return fmt.Errorf("query result exceeded memory threshold")
+			}
+
 			// Return response to the channel.
 			select {
 			case <-done:
@@ -5801,15 +5842,42 @@ func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapRe
 			}
 			return nil
 		})
+		if reterr != nil {
+			return reterr // exit early if error occurs when running serially
+		}
 	}
 	return nil
 }
 
+// calcResultMemory recursively computes the total memory used by v.
+func calcResultMemory(v interface{}) (n int64) {
+	switch v := v.(type) {
+	case string:
+	case ExtractedIDColumn:
+		n += 8 // ColumnID
+		for _, row := range v.Rows {
+			n += 24 + int64(len(row)*8) // slice header + data
+		}
+	case ExtractedIDMatrix:
+		n += 24 // slice size
+		for _, field := range v.Fields {
+			n += 16 + int64(len(field)) // string header + data
+		}
+
+		n += 24 // Columns slice
+		for _, col := range v.Columns {
+			n += calcResultMemory(col)
+		}
+	}
+	return n
+}
+
 type job struct {
-	shard      uint64
-	mapFn      mapFunc
-	ctx        context.Context
-	resultChan chan mapResponse
+	shard           uint64
+	mapFn           mapFunc
+	ctx             context.Context
+	memoryAvailable *int64 // shared, atomic value
+	resultChan      chan mapResponse
 }
 
 func worker(work chan job) {
@@ -5821,7 +5889,7 @@ func worker(work chan job) {
 			j.resultChan <- mapResponse{result: nil, err: err}
 			continue
 		}
-		result, err := j.mapFn(j.ctx, j.shard)
+		result, err := j.mapFn(j.ctx, j.shard, &mapOptions{memoryAvailable: j.memoryAvailable})
 		j.resultChan <- mapResponse{result: result, err: err}
 	}
 }
@@ -5829,7 +5897,7 @@ func worker(work chan job) {
 var errShutdown = errors.New("executor has shut down")
 
 // mapperLocal performs map & reduce entirely on the local node.
-func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFunc, reduceFn reduceFunc) (_ interface{}, err error) {
+func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFunc, reduceFn reduceFunc, memoryAvailable int64) (_ interface{}, err error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapperLocal")
 	defer span.Finish()
 	ctx, cancel := context.WithCancel(ctx)
@@ -5847,10 +5915,11 @@ func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFu
 	expected := 0
 	for _, shard := range shards {
 		j := job{
-			shard:      shard,
-			mapFn:      mapFn,
-			ctx:        ctx,
-			resultChan: ch,
+			shard:           shard,
+			mapFn:           mapFn,
+			ctx:             ctx,
+			resultChan:      ch,
+			memoryAvailable: &memoryAvailable,
 		}
 		select {
 		case <-done:
@@ -7175,7 +7244,11 @@ func validateQueryContext(ctx context.Context) error {
 // errShardUnavailable is a marker error if no nodes are available.
 var errShardUnavailable = errors.New("shard unavailable")
 
-type mapFunc func(ctx context.Context, shard uint64) (_ interface{}, err error)
+type mapFunc func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error)
+
+type mapOptions struct {
+	memoryAvailable *int64
+}
 
 type reduceFunc func(ctx context.Context, prev, v interface{}) interface{}
 
@@ -7193,6 +7266,7 @@ type execOptions struct {
 	Profile       bool
 	PreTranslated bool
 	EmbeddedData  []*Row
+	MaxMemory     int64
 }
 
 func needsShards(calls []*pql.Call) bool {
@@ -7983,7 +8057,7 @@ func (e *executor) executeDeleteRecords(ctx context.Context, qcx *Qcx, index str
 	}
 
 	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64) (_ interface{}, err error) {
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
 		return e.executeDeleteRecordFromShard(ctx, qcx, index, c, shard)
 	}
 
diff --git a/executor_test.go b/executor_test.go
index 23cdf4b4d..834906864 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -4925,6 +4925,76 @@ func TestExecutor_Execute_Extract_Keyed(t *testing.T) {
 	}
 }
 
+func TestExecutor_Execute_MaxMemory(t *testing.T) {
+	c := test.MustRunCluster(t, 3)
+	defer c.Close()
+
+	c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "set")
+	c.ImportBits(t, "i", "set", [][2]uint64{
+		{0, 1},
+		{0, 2},
+		{3, 1},
+		{4, 1},
+		{4, 4 * ShardWidth},
+		{5, ShardWidth},
+	})
+	c.Query(t, "i", fmt.Sprintf("Clear(%d, set=5)", ShardWidth))
+
+	resp := c.GetPrimary().QueryAPI(t, &pilosa.QueryRequest{
+		Index:     "i",
+		Query:     `Extract(All(), Rows(set))`,
+		MaxMemory: 1000,
+	})
+	expect := []interface{}{
+		pilosa.ExtractedTable{
+			Fields: []pilosa.ExtractedTableField{
+				{
+					Name: "set",
+					Type: "[]uint64",
+				},
+			},
+			Columns: []pilosa.ExtractedTableColumn{
+				{
+					Column: pilosa.KeyOrID{ID: 1},
+					Rows: []interface{}{
+						[]uint64{
+							0,
+							3,
+							4,
+						},
+					},
+				},
+				{
+					Column: pilosa.KeyOrID{ID: 2},
+					Rows: []interface{}{
+						[]uint64{
+							0,
+						},
+					},
+				},
+				{
+					Column: pilosa.KeyOrID{ID: ShardWidth},
+					Rows: []interface{}{
+						[]uint64{},
+					},
+				},
+				{
+					Column: pilosa.KeyOrID{ID: 4 * ShardWidth},
+					Rows: []interface{}{
+						[]uint64{
+							4,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	if !reflect.DeepEqual(expect, resp.Results) {
+		t.Errorf("expected %v but got %v", expect, resp.Results)
+	}
+}
+
 func TestExecutor_Execute_Rows(t *testing.T) {
 	c := test.MustRunCluster(t, 3)
 	defer c.Close()
diff --git a/handler.go b/handler.go
index eff3e3a37..49c4b1ff5 100644
--- a/handler.go
+++ b/handler.go
@@ -51,6 +51,9 @@ type QueryRequest struct {
 	// Additional data associated with the query, in cases where there's
 	// row-style inputs for precomputed values.
 	EmbeddedData []*Row
+
+	// Limit on memory used by request (Extract() only)
+	MaxMemory int64
 }
 
 // QueryResponse represent a response from a processed query.
diff --git a/pb/public.pb.go b/pb/public.pb.go
index 866040770..43ebfc641 100644
--- a/pb/public.pb.go
+++ b/pb/public.pb.go
@@ -1295,6 +1295,7 @@ type QueryRequest struct {
 	Remote               bool     `protobuf:"varint,5,opt,name=Remote,proto3" json:"Remote,omitempty"`
 	EmbeddedData         []*Row   `protobuf:"bytes,8,rep,name=EmbeddedData,proto3" json:"EmbeddedData,omitempty"`
 	PreTranslated        bool     `protobuf:"varint,9,opt,name=PreTranslated,proto3" json:"PreTranslated,omitempty"`
+	MaxMemory            int64    `protobuf:"varint,10,opt,name=MaxMemory,proto3" json:"MaxMemory,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -1368,6 +1369,13 @@ func (m *QueryRequest) GetPreTranslated() bool {
 	return false
 }
 
+func (m *QueryRequest) GetMaxMemory() int64 {
+	if m != nil {
+		return m.MaxMemory
+	}
+	return 0
+}
+
 type QueryResponse struct {
 	Err                  string         `protobuf:"bytes,1,opt,name=Err,proto3" json:"Err,omitempty"`
 	Results              []*QueryResult `protobuf:"bytes,2,rep,name=Results,proto3" json:"Results,omitempty"`
@@ -2441,104 +2449,106 @@ func init() {
 func init() { proto.RegisterFile("public.proto", fileDescriptor_413a91106d7bcce8) }
 
 var fileDescriptor_413a91106d7bcce8 = []byte{
-	// 1551 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x58, 0xdd, 0x6e, 0x1b, 0xc5,
-	0x17, 0xcf, 0x7a, 0xd7, 0x5f, 0xc7, 0x8e, 0x93, 0x4e, 0xd3, 0xfe, 0xf7, 0x5f, 0x52, 0xe3, 0xae,
-	0x50, 0xe5, 0x12, 0x94, 0x8a, 0x00, 0x15, 0xaa, 0x04, 0x28, 0x8e, 0x53, 0xb2, 0x6a, 0x9b, 0x96,
-	0x49, 0x08, 0x5c, 0x70, 0xb3, 0xb1, 0x07, 0x77, 0xc5, 0xda, 0x6b, 0xd6, 0xeb, 0x3a, 0x11, 0x0f,
-	0x00, 0x8f, 0xc0, 0x0b, 0x20, 0x1e, 0x05, 0xee, 0xe0, 0x92, 0x4b, 0x54, 0x5e, 0x04, 0x9d, 0x33,
-	0xb3, 0xbb, 0xb3, 0x6b, 0xb7, 0xaa, 0x2a, 0xee, 0xf6, 0x7c, 0xcc, 0xf9, 0xf8, 0xcd, 0xf9, 0x18,
-	0x1b, 0x9a, 0xd3, 0xf9, 0x79, 0xe0, 0x0f, 0x76, 0xa7, 0x51, 0x18, 0x87, 0xac, 0x34, 0x3d, 0x77,
-	0x2e, 0xc1, 0xe4, 0xe1, 0x82, 0xd9, 0x50, 0x3d, 0x08, 0x83, 0xf9, 0x78, 0x32, 0xb3, 0x8d, 0x8e,
-	0xd9, 0xb5, 0x78, 0x42, 0x32, 0x06, 0xd6, 0x43, 0x71, 0x39, 0xb3, 0xcd, 0x8e, 0xd9, 0xad, 0x73,
-	0xfa, 0x46, 0x6d, 0x1e, 0x7a, 0x91, 0x3f, 0x19, 0xd9, 0x56, 0xc7, 0xe8, 0x36, 0x79, 0x42, 0xb2,
-	0x2d, 0x28, 0xbb, 0x93, 0xa1, 0xb8, 0xb0, 0xcb, 0x1d, 0xa3, 0x5b, 0xe7, 0x92, 0x40, 0xee, 0x03,
-	0x5f, 0x04, 0x43, 0xbb, 0x22, 0xb9, 0x44, 0x38, 0x5d, 0xa8, 0xf3, 0x70, 0xf1, 0xd8, 0x8b, 0x23,
-	0xff, 0x82, 0xbd, 0x05, 0x16, 0x0f, 0x17, 0xd2, 0x7b, 0x63, 0xaf, 0xba, 0x3b, 0x3d, 0xdf, 0xe5,
-	0xe1, 0x82, 0x13, 0xd3, 0xd9, 0x87, 0xfa, 0x89, 0x3f, 0x9a, 0x88, 0x21, 0x86, 0xfa, 0x7f, 0x30,
-	0x9f, 0x86, 0xa8, 0x68, 0xe8, 0x8a, 0xc8, 0x43, 0xd1, 0xb1, 0x18, 0xd9, 0xa5, 0x82, 0xe8, 0x58,
-	0x8c, 0x9c, 0x8f, 0xa1, 0xc5, 0xc3, 0x85, 0x3b, 0x14, 0x93, 0xd8, 0xff, 0xd6, 0x17, 0x11, 0x25,
-	0x96, 0x7a, 0xb4, 0xa4, 0xa3, 0x34, 0xd9, 0x52, 0x96, 0xac, 0x73, 0x03, 0x2a, 0x6e, 0xff, 0x91,
-	0x3f, 0x8b, 0xd9, 0x26, 0x98, 0x6e, 0x3f, 0x39, 0x80, 0x9f, 0xce, 0x01, 0x5c, 0x39, 0xbc, 0x88,
-	0x23, 0x6f, 0x10, 0x8b, 0xa1, 0xdb, 0x97, 0x90, 0xb1, 0x16, 0x94, 0xdc, 0x3e, 0xc5, 0x67, 0xf1,
-	0x92, 0xdb, 0x67, 0x6d, 0xb0, 0xce, 0xbc, 0x40, 0x1a, 0x6d, 0xec, 0x01, 0x86, 0x25, 0x0d, 0x72,
-	0xe2, 0x3b, 0xdf, 0xe4, 0x8c, 0x28, 0x3c, 0xae, 0x43, 0x85, 0x50, 0x92, 0xee, 0xea, 0x5c, 0x51,
-	0xec, 0x6e, 0x76, 0x51, 0xd2, 0xde, 0x35, 0xb4, 0xb7, 0x14, 0x44, 0x7a, 0x7f, 0xce, 0x4d, 0xa8,
-	0x3e, 0x14, 0x97, 0x14, 0x7f, 0x92, 0x9d, 0xa1, 0x65, 0xf7, 0x87, 0x01, 0x57, 0xd3, 0xd3, 0xa7,
-	0xde, 0x79, 0x20, 0xce, 0xbc, 0x60, 0x2e, 0x58, 0x3b, 0xc9, 0xd5, 0xc8, 0xc7, 0x7c, 0xb4, 0x46,
-	0x99, 0xb3, 0x5b, 0x29, 0x52, 0xa8, 0xd0, 0x40, 0x05, 0xe5, 0xe6, 0x68, 0x4d, 0x55, 0xc9, 0x36,
-	0xd4, 0x7a, 0x27, 0x2e, 0x99, 0xb3, 0xcd, 0x8e, 0xd1, 0x35, 0x8f, 0xd6, 0x78, 0xca, 0x61, 0x37,
-	0xa0, 0xfa, 0x78, 0x1e, 0x8b, 0x0b, 0xb7, 0x4f, 0x35, 0x64, 0x1d, 0xad, 0xf1, 0x84, 0x81, 0x27,
-	0xe9, 0xf3, 0xa1, 0xb8, 0x94, 0x85, 0x84, 0x27, 0x13, 0x0e, 0xdb, 0x02, 0xab, 0x17, 0x86, 0x01,
-	0x15, 0x53, 0x0d, 0xbd, 0x21, 0xd5, 0xab, 0x42, 0x99, 0x0c, 0x3b, 0x17, 0xb0, 0x95, 0x4f, 0x48,
-	0x5d, 0x0b, 0x03, 0x13, 0xed, 0x19, 0xca, 0x1e, 0x12, 0x6c, 0x93, 0xae, 0xaa, 0xa4, 0xfc, 0xe3,
-	0x65, 0xdd, 0x85, 0x0a, 0x99, 0x91, 0x05, 0xdf, 0xd8, 0xfb, 0x5f, 0x0e, 0xde, 0x0c, 0x20, 0xae,
-	0xd4, 0x7a, 0x75, 0xc2, 0xf7, 0x49, 0xe4, 0xf6, 0x9d, 0x4f, 0x8a, 0x50, 0xd2, 0x9d, 0x21, 0xec,
-	0xc7, 0xde, 0x58, 0x48, 0xcf, 0x9c, 0xbe, 0x91, 0x77, 0x7a, 0x39, 0x15, 0xe4, 0xba, 0xce, 0xe9,
-	0xdb, 0x99, 0x43, 0x2b, 0x7f, 0x1c, 0x83, 0xd1, 0x8a, 0x60, 0x65, 0x30, 0x24, 0x4f, 0xab, 0x63,
-	0xaf, 0x58, 0x1d, 0xf6, 0xf2, 0x89, 0x62, 0x81, 0x7c, 0x0a, 0xd6, 0x53, 0xcf, 0x8f, 0x96, 0xca,
-	0x76, 0x53, 0xe2, 0x65, 0x52, 0x84, 0xa6, 0x04, 0xbe, 0x7c, 0x10, 0xce, 0x27, 0xb1, 0x04, 0x8c,
-	0x4b, 0xc2, 0xf9, 0x0c, 0xea, 0x78, 0x5e, 0xe6, 0xba, 0x2d, 0x8d, 0xa9, 0xba, 0xa9, 0xa1, 0x77,
-	0xa4, 0xb9, 0x74, 0x91, 0xce, 0x81, 0x92, 0x3e, 0x07, 0x7a, 0x00, 0x28, 0x9d, 0x49, 0x0b, 0x6d,
-	0x28, 0x13, 0xa5, 0x52, 0xce, 0x4c, 0x48, 0xf6, 0x4b, 0x6c, 0xdc, 0xc4, 0xb9, 0x13, 0xdf, 0xfb,
-	0x10, 0xc5, 0xb2, 0xe2, 0x30, 0x02, 0x93, 0xab, 0x9a, 0x08, 0xa1, 0x26, 0x81, 0x0a, 0x17, 0x99,
-	0x01, 0x43, 0x33, 0x80, 0x5c, 0x9c, 0x0f, 0xfd, 0x24, 0x37, 0x22, 0xb0, 0x0b, 0x79, 0xb8, 0xc8,
-	0x60, 0x50, 0x14, 0x7b, 0x3b, 0xf1, 0x62, 0x51, 0x9e, 0x75, 0xea, 0x0f, 0xf4, 0x9f, 0x38, 0xfc,
-	0x1a, 0xe0, 0xf3, 0x28, 0x9c, 0x4f, 0x09, 0x22, 0xe6, 0x40, 0x99, 0x28, 0x95, 0x53, 0x13, 0xd5,
-	0x93, 0x78, 0xb8, 0x14, 0xad, 0x06, 0x17, 0x2f, 0x61, 0x7f, 0x34, 0x92, 0xed, 0xc3, 0xf1, 0xd3,
-	0xf9, 0x01, 0x6a, 0x67, 0x5e, 0x90, 0x4a, 0xcf, 0xbc, 0x40, 0xa5, 0x8a, 0x9f, 0x79, 0x2b, 0x66,
-	0x62, 0xe5, 0x06, 0xd4, 0x1e, 0x04, 0xa1, 0x17, 0xa3, 0x32, 0x9a, 0x32, 0x78, 0x4a, 0xb3, 0x1d,
-	0x80, 0xbe, 0x18, 0xf8, 0x63, 0x2f, 0x40, 0xa9, 0x95, 0xb5, 0xb3, 0xe2, 0x72, 0x4d, 0xec, 0x7c,
-	0x04, 0x55, 0x45, 0xad, 0x06, 0x1a, 0xb9, 0x27, 0x03, 0x2f, 0x10, 0x89, 0x7f, 0x22, 0x9c, 0x5f,
-	0x0d, 0x68, 0x7e, 0x31, 0x17, 0xd1, 0x25, 0x17, 0xdf, 0xcf, 0xc5, 0x2c, 0x46, 0x35, 0xa2, 0x93,
-	0x3b, 0x20, 0x02, 0xd1, 0x3e, 0x79, 0xe6, 0x45, 0x43, 0x59, 0xbc, 0x16, 0x57, 0x14, 0xdd, 0x82,
-	0x18, 0x87, 0xb1, 0xa0, 0x61, 0x50, 0xe3, 0x8a, 0x62, 0x3b, 0xd0, 0x3c, 0x1c, 0x9f, 0x8b, 0xe1,
-	0x50, 0x0c, 0xfb, 0x5e, 0xec, 0xd9, 0xb5, 0xfc, 0xee, 0xc8, 0x09, 0xd9, 0x3b, 0xb0, 0xfe, 0x34,
-	0x12, 0xa7, 0x91, 0x37, 0x99, 0x05, 0x5e, 0x2c, 0x86, 0x76, 0x9d, 0x6c, 0xe5, 0x99, 0xce, 0x23,
-	0x58, 0x57, 0x81, 0xce, 0xa6, 0xe1, 0x64, 0x26, 0x10, 0xe2, 0xc3, 0x28, 0x52, 0x71, 0xe2, 0x27,
-	0xbb, 0x03, 0x55, 0x2e, 0x66, 0xf3, 0x20, 0x4e, 0x7a, 0x6c, 0x03, 0x1d, 0x26, 0xa7, 0xe6, 0x41,
-	0xcc, 0x13, 0xb9, 0xf3, 0x4b, 0x19, 0x1a, 0x9a, 0x20, 0xed, 0x7a, 0x9c, 0x5c, 0xeb, 0xb2, 0xeb,
-	0x71, 0x67, 0xf1, 0x70, 0xb1, 0xb4, 0xce, 0xb0, 0x52, 0x9b, 0x60, 0x1c, 0xab, 0x72, 0x30, 0x8e,
-	0xb3, 0xc6, 0x30, 0x57, 0x37, 0x06, 0xae, 0xf0, 0x67, 0xde, 0x64, 0x24, 0x86, 0x74, 0x8b, 0x35,
-	0x9e, 0x90, 0xac, 0x9b, 0x95, 0x0c, 0x21, 0xa8, 0x2a, 0x30, 0xe1, 0xf1, 0xac, 0xa0, 0x64, 0xbd,
-	0xe3, 0xe0, 0xaf, 0xca, 0x1b, 0x90, 0x14, 0xbb, 0x07, 0xad, 0x27, 0xc1, 0x30, 0xab, 0xe8, 0x99,
-	0xc2, 0xba, 0x85, 0x76, 0x32, 0x36, 0x2f, 0x68, 0xb1, 0xfb, 0xc5, 0xad, 0x4b, 0xa8, 0x37, 0xf6,
-	0x98, 0xca, 0x53, 0x93, 0xf0, 0xe2, 0x7e, 0xde, 0xd1, 0x96, 0xbe, 0x0d, 0x74, 0x6c, 0x1d, 0x8f,
-	0xa5, 0x4c, 0xae, 0x3d, 0x0a, 0x76, 0xf5, 0x19, 0x62, 0x37, 0x48, 0xbb, 0x95, 0x20, 0x24, 0xb9,
-	0x5c, 0x9f, 0x32, 0x3b, 0xda, 0xd0, 0xb2, 0x9b, 0x99, 0xf1, 0x94, 0xc9, 0xb5, 0xa1, 0x76, 0xb0,
-	0x62, 0x41, 0xdb, 0xeb, 0x74, 0xa8, 0xb8, 0x7d, 0xa5, 0x90, 0xaf, 0x58, 0xe8, 0xf7, 0x8b, 0xd3,
-	0xdd, 0x6e, 0x65, 0x50, 0xe4, 0x25, 0xbc, 0xb8, 0x07, 0x76, 0xb4, 0x97, 0x92, 0xbd, 0x91, 0x45,
-	0x9b, 0x32, 0xb9, 0xf6, 0x92, 0x7a, 0x1f, 0x1a, 0xfa, 0x45, 0x6d, 0x92, 0xfa, 0x46, 0xfe, 0xa2,
-	0x66, 0x5c, 0xd7, 0x71, 0x7e, 0x2b, 0xc1, 0xba, 0x3b, 0x9e, 0x86, 0x51, 0xac, 0x35, 0xa8, 0x7c,
-	0xc7, 0x19, 0x2b, 0xdf, 0x71, 0xa5, 0xc2, 0xe8, 0xa4, 0x46, 0xa5, 0xd1, 0x62, 0x71, 0x49, 0x68,
-	0xa5, 0x64, 0xe5, 0x4a, 0x69, 0x1b, 0xea, 0x72, 0xf3, 0xa0, 0xa8, 0x4c, 0xa2, 0x8c, 0x21, 0x5f,
-	0x96, 0x0b, 0x7a, 0x59, 0x54, 0xe9, 0x95, 0x92, 0x90, 0xac, 0x0d, 0x20, 0xd5, 0x48, 0x58, 0x23,
-	0xa1, 0xc6, 0x41, 0xf9, 0xa9, 0x3f, 0x16, 0xb3, 0xd8, 0x1b, 0x4f, 0x67, 0x76, 0xa5, 0x63, 0x76,
-	0x4d, 0xae, 0x71, 0xd8, 0x6d, 0x68, 0x51, 0x12, 0x07, 0x91, 0xc0, 0x4e, 0xdf, 0x8f, 0xa9, 0x14,
-	0x4d, 0x5e, 0xe0, 0xa2, 0x1e, 0xa5, 0x95, 0xe9, 0x81, 0xd4, 0xcb, 0x73, 0x69, 0xd2, 0x06, 0xc2,
-	0x8b, 0xa8, 0xd8, 0x6a, 0x5c, 0x12, 0xce, 0x5f, 0x25, 0x60, 0x12, 0x49, 0xf9, 0x4a, 0xf8, 0xcf,
-	0xe0, 0x7c, 0x35, 0x6c, 0x79, 0x70, 0xaa, 0x4b, 0xe0, 0x5c, 0x4f, 0x5f, 0x35, 0x12, 0x18, 0x45,
-	0xb1, 0x0e, 0x34, 0x92, 0x45, 0x80, 0x42, 0x44, 0xd5, 0xe0, 0x3a, 0x8b, 0x39, 0xd0, 0x3c, 0x89,
-	0xf1, 0x69, 0xaf, 0x54, 0xea, 0x64, 0x3b, 0xc7, 0x5b, 0x01, 0x2d, 0xbc, 0x26, 0xb4, 0x8d, 0x57,
-	0x43, 0xdb, 0xd4, 0xa1, 0xfd, 0xd1, 0x80, 0xe6, 0x7e, 0x1c, 0x8e, 0xfd, 0x01, 0x17, 0x83, 0x30,
-	0x1a, 0xbe, 0x1c, 0x54, 0x09, 0x5f, 0x49, 0x87, 0xaf, 0x0b, 0xa6, 0xfb, 0x3c, 0x52, 0xa3, 0xf3,
-	0x3a, 0xad, 0xeb, 0xa5, 0x5b, 0xe2, 0xa8, 0xc2, 0x6e, 0x41, 0xc9, 0x8d, 0xa8, 0x66, 0x1b, 0x7b,
-	0x57, 0x32, 0xc5, 0x44, 0xa7, 0xe4, 0x46, 0xce, 0x7b, 0xb0, 0x25, 0x03, 0x49, 0x44, 0x6a, 0x57,
-	0x6c, 0x41, 0xf9, 0x30, 0x8a, 0xc2, 0x64, 0x5b, 0x48, 0x02, 0xdf, 0xa3, 0xe9, 0x82, 0xc1, 0xcb,
-	0x78, 0x93, 0x9a, 0x58, 0xf5, 0x23, 0xac, 0x03, 0x8d, 0xe3, 0x30, 0xfe, 0x2a, 0xf2, 0x63, 0x9a,
-	0x26, 0x72, 0xe6, 0xeb, 0x2c, 0xe7, 0x0e, 0x5c, 0x2b, 0x78, 0xce, 0x96, 0x1a, 0x96, 0x91, 0x99,
-	0xfd, 0x90, 0x39, 0x81, 0xab, 0xa9, 0xaa, 0xdb, 0x7f, 0xa3, 0x18, 0x97, 0x8d, 0xbe, 0xab, 0x65,
-	0x4e, 0x46, 0x95, 0xfb, 0x15, 0xd9, 0x38, 0x3d, 0xb0, 0x15, 0x9a, 0xf2, 0x97, 0xa4, 0x8a, 0xe0,
-	0xcc, 0x17, 0x8b, 0x97, 0x3d, 0xa0, 0x69, 0xe7, 0x97, 0xe8, 0xf7, 0x27, 0x7d, 0x3b, 0x3f, 0x95,
-	0x60, 0x6b, 0x95, 0x91, 0xac, 0xa0, 0x0c, 0xad, 0xa0, 0xd8, 0x1e, 0x94, 0x9f, 0xfb, 0x62, 0x91,
-	0xac, 0xf1, 0x6d, 0xed, 0xb2, 0x97, 0x62, 0xe0, 0x52, 0x15, 0x1b, 0x69, 0x7f, 0x10, 0xfb, 0xe1,
-	0x24, 0x79, 0x10, 0x4a, 0x0a, 0x3d, 0xf4, 0x82, 0x70, 0xf0, 0x9d, 0xfc, 0x2d, 0xc3, 0x25, 0xb1,
-	0xa2, 0x31, 0xca, 0xaf, 0xd9, 0x18, 0x95, 0x95, 0x8d, 0xd1, 0x85, 0x8d, 0x2f, 0xa7, 0x43, 0x2f,
-	0x16, 0x87, 0x17, 0xfe, 0x2c, 0x16, 0x93, 0x81, 0xb0, 0xab, 0x94, 0x51, 0x91, 0xed, 0x9c, 0xe4,
-	0x96, 0x00, 0x4e, 0x8f, 0xfd, 0xd1, 0x28, 0x12, 0x23, 0x2f, 0x4e, 0x60, 0xcc, 0x18, 0xec, 0x36,
-	0x54, 0x48, 0x39, 0x41, 0xa2, 0xb8, 0xd5, 0x95, 0xb4, 0xb7, 0xf9, 0xfb, 0x8b, 0xb6, 0xf1, 0xe7,
-	0x8b, 0xb6, 0xf1, 0xf7, 0x8b, 0xb6, 0xf1, 0xf3, 0x3f, 0xed, 0xb5, 0xf3, 0x0a, 0xfd, 0x91, 0xf0,
-	0xc1, 0xbf, 0x01, 0x00, 0x00, 0xff, 0xff, 0x3f, 0xf7, 0x85, 0x80, 0x58, 0x10, 0x00, 0x00,
+	// 1569 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x58, 0xdd, 0x6e, 0x1b, 0xd5,
+	0x16, 0xce, 0xfc, 0xf8, 0x6f, 0xd9, 0x71, 0xd2, 0xdd, 0xb4, 0x67, 0x4e, 0x4f, 0xea, 0xe3, 0x8e,
+	0x8e, 0x2a, 0xf7, 0x04, 0xa5, 0xc2, 0x40, 0x85, 0x2a, 0x01, 0x8a, 0xe3, 0x94, 0x8c, 0xda, 0xa4,
+	0x65, 0x27, 0x04, 0x2e, 0xb8, 0x99, 0xd8, 0x1b, 0x77, 0xc4, 0xd8, 0x63, 0xc6, 0xe3, 0x3a, 0x16,
+	0x0f, 0x00, 0x8f, 0xc0, 0x0b, 0xf0, 0x28, 0x08, 0xee, 0xe0, 0x92, 0x4b, 0x54, 0x5e, 0x04, 0xad,
+	0xb5, 0xf7, 0xfc, 0xda, 0xad, 0xaa, 0x8a, 0xbb, 0x59, 0x3f, 0x7b, 0xfd, 0x7c, 0x6b, 0xed, 0xb5,
+	0xb6, 0x0d, 0x8d, 0xe9, 0xfc, 0xd2, 0xf7, 0x06, 0xfb, 0xd3, 0x30, 0x88, 0x02, 0xa6, 0x4f, 0x2f,
+	0xed, 0x25, 0x18, 0x3c, 0x58, 0x30, 0x0b, 0x2a, 0x87, 0x81, 0x3f, 0x1f, 0x4f, 0x66, 0x96, 0xd6,
+	0x36, 0x3a, 0x26, 0x8f, 0x49, 0xc6, 0xc0, 0x7c, 0x2c, 0x96, 0x33, 0xcb, 0x68, 0x1b, 0x9d, 0x1a,
+	0xa7, 0x6f, 0xd4, 0xe6, 0x81, 0x1b, 0x7a, 0x93, 0x91, 0x65, 0xb6, 0xb5, 0x4e, 0x83, 0xc7, 0x24,
+	0xdb, 0x81, 0x92, 0x33, 0x19, 0x8a, 0x2b, 0xab, 0xd4, 0xd6, 0x3a, 0x35, 0x2e, 0x09, 0xe4, 0x3e,
+	0xf2, 0x84, 0x3f, 0xb4, 0xca, 0x92, 0x4b, 0x84, 0xdd, 0x81, 0x1a, 0x0f, 0x16, 0x27, 0x6e, 0x14,
+	0x7a, 0x57, 0xec, 0x3f, 0x60, 0xf2, 0x60, 0x21, 0xbd, 0xd7, 0xbb, 0x95, 0xfd, 0xe9, 0xe5, 0x3e,
+	0x0f, 0x16, 0x9c, 0x98, 0xf6, 0x01, 0xd4, 0xce, 0xbc, 0xd1, 0x44, 0x0c, 0x31, 0xd4, 0x7f, 0x83,
+	0xf1, 0x2c, 0x40, 0x45, 0x2d, 0xab, 0x88, 0x3c, 0x14, 0x9d, 0x8a, 0x91, 0xa5, 0x17, 0x44, 0xa7,
+	0x62, 0x64, 0x7f, 0x08, 0x4d, 0x1e, 0x2c, 0x9c, 0xa1, 0x98, 0x44, 0xde, 0xd7, 0x9e, 0x08, 0x29,
+	0xb1, 0xc4, 0xa3, 0x29, 0x1d, 0x25, 0xc9, 0xea, 0x69, 0xb2, 0xf6, 0x2d, 0x28, 0x3b, 0xfd, 0x27,
+	0xde, 0x2c, 0x62, 0xdb, 0x60, 0x38, 0xfd, 0xf8, 0x00, 0x7e, 0xda, 0x87, 0x70, 0xed, 0xe8, 0x2a,
+	0x0a, 0xdd, 0x41, 0x24, 0x86, 0x4e, 0x5f, 0x42, 0xc6, 0x9a, 0xa0, 0x3b, 0x7d, 0x8a, 0xcf, 0xe4,
+	0xba, 0xd3, 0x67, 0x2d, 0x30, 0x2f, 0x5c, 0x5f, 0x1a, 0xad, 0x77, 0x01, 0xc3, 0x92, 0x06, 0x39,
+	0xf1, 0xed, 0xaf, 0x72, 0x46, 0x14, 0x1e, 0x37, 0xa1, 0x4c, 0x28, 0x49, 0x77, 0x35, 0xae, 0x28,
+	0x76, 0x3f, 0x2d, 0x94, 0xb4, 0x77, 0x03, 0xed, 0xad, 0x04, 0x91, 0xd4, 0xcf, 0xbe, 0x0d, 0x95,
+	0xc7, 0x62, 0x49, 0xf1, 0xc7, 0xd9, 0x69, 0x99, 0xec, 0x7e, 0xd3, 0xe0, 0x7a, 0x72, 0xfa, 0xdc,
+	0xbd, 0xf4, 0xc5, 0x85, 0xeb, 0xcf, 0x05, 0x6b, 0xc5, 0xb9, 0x6a, 0xf9, 0x98, 0x8f, 0x37, 0x28,
+	0x73, 0x76, 0x27, 0x41, 0x0a, 0x15, 0xea, 0xa8, 0xa0, 0xdc, 0x1c, 0x6f, 0xa8, 0x2e, 0xd9, 0x85,
+	0x6a, 0xef, 0xcc, 0x21, 0x73, 0x96, 0xd1, 0xd6, 0x3a, 0xc6, 0xf1, 0x06, 0x4f, 0x38, 0xec, 0x16,
+	0x54, 0x4e, 0xe6, 0x91, 0xb8, 0x72, 0xfa, 0xd4, 0x43, 0xe6, 0xf1, 0x06, 0x8f, 0x19, 0x78, 0x92,
+	0x3e, 0x1f, 0x8b, 0xa5, 0x6c, 0x24, 0x3c, 0x19, 0x73, 0xd8, 0x0e, 0x98, 0xbd, 0x20, 0xf0, 0xa9,
+	0x99, 0xaa, 0xe8, 0x0d, 0xa9, 0x5e, 0x05, 0x4a, 0x64, 0xd8, 0xbe, 0x82, 0x9d, 0x7c, 0x42, 0xaa,
+	0x2c, 0x0c, 0x0c, 0xb4, 0xa7, 0x29, 0x7b, 0x48, 0xb0, 0x6d, 0x2a, 0x95, 0xae, 0xfc, 0x63, 0xb1,
+	0xee, 0x43, 0x99, 0xcc, 0xc8, 0x86, 0xaf, 0x77, 0xff, 0x95, 0x83, 0x37, 0x05, 0x88, 0x2b, 0xb5,
+	0x5e, 0x8d, 0xf0, 0x7d, 0x1a, 0x3a, 0x7d, 0xfb, 0xa3, 0x22, 0x94, 0x54, 0x33, 0x84, 0xfd, 0xd4,
+	0x1d, 0x0b, 0xe9, 0x99, 0xd3, 0x37, 0xf2, 0xce, 0x97, 0x53, 0x41, 0xae, 0x6b, 0x9c, 0xbe, 0xed,
+	0x39, 0x34, 0xf3, 0xc7, 0x31, 0x98, 0x4c, 0x13, 0xac, 0x0d, 0x86, 0xe4, 0x49, 0x77, 0x74, 0x8b,
+	0xdd, 0x61, 0xad, 0x9e, 0x28, 0x36, 0xc8, 0xc7, 0x60, 0x3e, 0x73, 0xbd, 0x70, 0xa5, 0x6d, 0xb7,
+	0x25, 0x5e, 0x06, 0x45, 0x68, 0x48, 0xe0, 0x4b, 0x87, 0xc1, 0x7c, 0x12, 0x49, 0xc0, 0xb8, 0x24,
+	0xec, 0x4f, 0xa0, 0x86, 0xe7, 0x65, 0xae, 0xbb, 0xd2, 0x98, 0xea, 0x9b, 0x2a, 0x7a, 0x47, 0x9a,
+	0x4b, 0x17, 0xc9, 0x1c, 0xd0, 0xb3, 0x73, 0xa0, 0x07, 0x80, 0xd2, 0x99, 0xb4, 0xd0, 0x82, 0x12,
+	0x51, 0x2a, 0xe5, 0xd4, 0x84, 0x64, 0xbf, 0xc2, 0xc6, 0x6d, 0x9c, 0x3b, 0xd1, 0x83, 0xf7, 0x51,
+	0x2c, 0x3b, 0x0e, 0x23, 0x30, 0xb8, 0xea, 0x89, 0x00, 0xaa, 0x12, 0xa8, 0x60, 0x91, 0x1a, 0xd0,
+	0x32, 0x06, 0x90, 0x8b, 0xf3, 0xa1, 0x1f, 0xe7, 0x46, 0x04, 0xde, 0x42, 0x1e, 0x2c, 0x52, 0x18,
+	0x14, 0xc5, 0xfe, 0x1b, 0x7b, 0x31, 0x29, 0xcf, 0x1a, 0xdd, 0x0f, 0xf4, 0x1f, 0x3b, 0xfc, 0x12,
+	0xe0, 0xd3, 0x30, 0x98, 0x4f, 0x09, 0x22, 0x66, 0x43, 0x89, 0x28, 0x95, 0x53, 0x03, 0xd5, 0xe3,
+	0x78, 0xb8, 0x14, 0xad, 0x07, 0x17, 0x8b, 0x70, 0x30, 0x1a, 0xc9, 0xeb, 0xc3, 0xf1, 0xd3, 0xfe,
+	0x0e, 0xaa, 0x17, 0xae, 0x9f, 0x48, 0x2f, 0x5c, 0x5f, 0xa5, 0x8a, 0x9f, 0x79, 0x2b, 0x46, 0x6c,
+	0xe5, 0x16, 0x54, 0x1f, 0xf9, 0x81, 0x1b, 0xa1, 0x32, 0x9a, 0xd2, 0x78, 0x42, 0xb3, 0x3d, 0x80,
+	0xbe, 0x18, 0x78, 0x63, 0xd7, 0x47, 0xa9, 0x99, 0x5e, 0x67, 0xc5, 0xe5, 0x19, 0xb1, 0xfd, 0x01,
+	0x54, 0x14, 0xb5, 0x1e, 0x68, 0xe4, 0x9e, 0x0d, 0x5c, 0x5f, 0xc4, 0xfe, 0x89, 0xb0, 0x7f, 0xd6,
+	0xa0, 0xf1, 0xd9, 0x5c, 0x84, 0x4b, 0x2e, 0xbe, 0x9d, 0x8b, 0x59, 0x84, 0x6a, 0x44, 0xc7, 0x35,
+	0x20, 0x02, 0xd1, 0x3e, 0x7b, 0xee, 0x86, 0x43, 0xd9, 0xbc, 0x26, 0x57, 0x14, 0x55, 0x41, 0x8c,
+	0x83, 0x48, 0xd0, 0x30, 0xa8, 0x72, 0x45, 0xb1, 0x3d, 0x68, 0x1c, 0x8d, 0x2f, 0xc5, 0x70, 0x28,
+	0x86, 0x7d, 0x37, 0x72, 0xad, 0x6a, 0x7e, 0x77, 0xe4, 0x84, 0xec, 0x7f, 0xb0, 0xf9, 0x2c, 0x14,
+	0xe7, 0xa1, 0x3b, 0x99, 0xf9, 0x6e, 0x24, 0x86, 0x56, 0x8d, 0x6c, 0xe5, 0x99, 0x6c, 0x17, 0x6a,
+	0x27, 0xee, 0xd5, 0x89, 0x18, 0x07, 0xe1, 0xd2, 0x02, 0xca, 0x21, 0x65, 0xd8, 0x4f, 0x60, 0x53,
+	0xa5, 0x31, 0x9b, 0x06, 0x93, 0x99, 0xc0, 0x02, 0x1c, 0x85, 0xa1, 0xca, 0x02, 0x3f, 0xd9, 0x3d,
+	0xa8, 0x70, 0x31, 0x9b, 0xfb, 0x51, 0x7c, 0x03, 0xb7, 0x30, 0x9c, 0xf8, 0xd4, 0xdc, 0x8f, 0x78,
+	0x2c, 0xb7, 0x7f, 0x2a, 0x41, 0x3d, 0x23, 0x48, 0x66, 0x02, 0xce, 0xb5, 0x4d, 0x39, 0x13, 0x70,
+	0xa3, 0xf1, 0x60, 0xb1, 0xb2, 0xec, 0xb0, 0x8f, 0x1b, 0xa0, 0x9d, 0xaa, 0x66, 0xd1, 0x4e, 0xd3,
+	0x6b, 0x63, 0xac, 0xbf, 0x36, 0xb8, 0xe0, 0x9f, 0xbb, 0x93, 0x91, 0x18, 0x52, 0x8d, 0xab, 0x3c,
+	0x26, 0x59, 0x27, 0x6d, 0x28, 0xc2, 0x57, 0xf5, 0x67, 0xcc, 0xe3, 0x69, 0xbb, 0xc9, 0xdb, 0x80,
+	0x6b, 0xa1, 0x22, 0xeb, 0x23, 0x29, 0xf6, 0x00, 0x9a, 0x4f, 0xfd, 0x61, 0xda, 0xef, 0x33, 0x55,
+	0x89, 0x26, 0xda, 0x49, 0xd9, 0xbc, 0xa0, 0xc5, 0x1e, 0x16, 0x77, 0x32, 0xd5, 0xa4, 0xde, 0x65,
+	0x2a, 0xcf, 0x8c, 0x84, 0x17, 0xb7, 0xf7, 0x5e, 0xe6, 0x49, 0x40, 0x85, 0xaa, 0x77, 0x37, 0xf1,
+	0x58, 0xc2, 0xe4, 0x99, 0x27, 0xc3, 0x7e, 0x76, 0xc2, 0x58, 0x75, 0xd2, 0x6e, 0xc6, 0x08, 0x49,
+	0x2e, 0xcf, 0xce, 0xa0, 0xbd, 0xcc, 0x48, 0xb3, 0x1a, 0xa9, 0xf1, 0x84, 0xc9, 0x33, 0x23, 0xef,
+	0x70, 0xcd, 0xfa, 0xb6, 0x36, 0xe9, 0x50, 0x71, 0x37, 0x4b, 0x21, 0x5f, 0xb3, 0xee, 0x1f, 0x16,
+	0x67, 0xbf, 0xd5, 0x4c, 0xa1, 0xc8, 0x4b, 0x78, 0x71, 0x4b, 0xec, 0x65, 0xde, 0x51, 0xd6, 0x56,
+	0x1a, 0x6d, 0xc2, 0xe4, 0x99, 0x77, 0xd6, 0xbb, 0x50, 0xcf, 0x16, 0x6a, 0x9b, 0xd4, 0xb7, 0xf2,
+	0x85, 0x9a, 0xf1, 0xac, 0x8e, 0xfd, 0x8b, 0x0e, 0x9b, 0xce, 0x78, 0x1a, 0x84, 0x51, 0xe6, 0xfa,
+	0xca, 0x57, 0x9e, 0xb6, 0xf6, 0x95, 0xa7, 0x17, 0x06, 0x2b, 0x5d, 0x63, 0x1a, 0x3c, 0x26, 0x97,
+	0x44, 0xa6, 0x95, 0xcc, 0x5c, 0x2b, 0xed, 0x42, 0x4d, 0xee, 0x25, 0x14, 0x95, 0x48, 0x94, 0x32,
+	0xe4, 0xbb, 0x73, 0x41, 0xef, 0x8e, 0x0a, 0xbd, 0x61, 0x62, 0x92, 0xb5, 0x00, 0xa4, 0x1a, 0x09,
+	0xab, 0x24, 0xcc, 0x70, 0x50, 0x7e, 0xee, 0x8d, 0xc5, 0x2c, 0x72, 0xc7, 0xd3, 0x99, 0x55, 0x6e,
+	0x1b, 0x1d, 0x83, 0x67, 0x38, 0xec, 0x2e, 0x34, 0x29, 0x89, 0xc3, 0x50, 0xe0, 0x1c, 0x38, 0x88,
+	0xa8, 0x15, 0x0d, 0x5e, 0xe0, 0xa2, 0x1e, 0xa5, 0x95, 0xea, 0xc9, 0x21, 0x51, 0xe0, 0xd2, 0x1c,
+	0xf6, 0x85, 0x1b, 0x52, 0xb3, 0x55, 0xb9, 0x24, 0xec, 0x3f, 0x74, 0x60, 0x12, 0x49, 0xf9, 0x86,
+	0xf8, 0xc7, 0xe0, 0x7c, 0x3d, 0x6c, 0x79, 0x70, 0x2a, 0x2b, 0xe0, 0xdc, 0x4c, 0xde, 0x3c, 0x12,
+	0x18, 0x45, 0xb1, 0x36, 0xd4, 0xe3, 0x35, 0x81, 0x42, 0x44, 0x55, 0xe3, 0x59, 0x16, 0xb3, 0xa1,
+	0x71, 0x16, 0xe1, 0xc3, 0x5f, 0xa9, 0xd4, 0xc8, 0x76, 0x8e, 0xb7, 0x06, 0x5a, 0x78, 0x43, 0x68,
+	0xeb, 0xaf, 0x87, 0xb6, 0x91, 0x85, 0xf6, 0x7b, 0x0d, 0x1a, 0x07, 0x51, 0x30, 0xf6, 0x06, 0x5c,
+	0x0c, 0x82, 0x70, 0xf8, 0x6a, 0x50, 0x25, 0x7c, 0x7a, 0x16, 0xbe, 0x0e, 0x18, 0xce, 0x8b, 0x50,
+	0x8d, 0xce, 0x9b, 0xb4, 0xcc, 0x57, 0xaa, 0xc4, 0x51, 0x85, 0xdd, 0x01, 0xdd, 0x09, 0xa9, 0x67,
+	0xeb, 0xdd, 0x6b, 0xa9, 0x62, 0xac, 0xa3, 0x3b, 0xa1, 0xfd, 0x0e, 0xec, 0xc8, 0x40, 0x62, 0x91,
+	0xda, 0x15, 0x3b, 0x50, 0x3a, 0x0a, 0xc3, 0x20, 0xde, 0x16, 0x92, 0xc0, 0xd7, 0x6a, 0xb2, 0x7e,
+	0xb0, 0x18, 0x6f, 0xd3, 0x13, 0xeb, 0x7e, 0xa2, 0xb5, 0xa1, 0x7e, 0x1a, 0x44, 0x5f, 0x84, 0x5e,
+	0x44, 0xd3, 0x44, 0xce, 0xfc, 0x2c, 0xcb, 0xbe, 0x07, 0x37, 0x0a, 0x9e, 0xd3, 0xa5, 0x86, 0x6d,
+	0x64, 0xa4, 0x3f, 0x73, 0xce, 0xe0, 0x7a, 0xa2, 0xea, 0xf4, 0xdf, 0x2a, 0xc6, 0x55, 0xa3, 0xff,
+	0xcf, 0x64, 0x4e, 0x46, 0x95, 0xfb, 0x35, 0xd9, 0xd8, 0x3d, 0xb0, 0x14, 0x9a, 0xf2, 0x77, 0xa6,
+	0x8a, 0xe0, 0xc2, 0x13, 0x8b, 0x57, 0x3d, 0xaf, 0xe9, 0x45, 0xa0, 0xd3, 0xaf, 0x53, 0xfa, 0xb6,
+	0x7f, 0xd0, 0x61, 0x67, 0x9d, 0x91, 0xb4, 0xa1, 0xb4, 0x4c, 0x43, 0xb1, 0x2e, 0x94, 0x5e, 0x78,
+	0x62, 0x11, 0xaf, 0xf1, 0xdd, 0x4c, 0xb1, 0x57, 0x62, 0xe0, 0x52, 0x15, 0x2f, 0xd2, 0xc1, 0x20,
+	0xf2, 0x82, 0x49, 0xfc, 0x5c, 0x94, 0x14, 0x7a, 0xe8, 0xf9, 0xc1, 0xe0, 0x1b, 0xf9, 0x4b, 0x87,
+	0x4b, 0x62, 0xcd, 0xc5, 0x28, 0xbd, 0xe1, 0xc5, 0x28, 0xaf, 0xbd, 0x18, 0x1d, 0xd8, 0xfa, 0x7c,
+	0x3a, 0x74, 0x23, 0x71, 0x74, 0xe5, 0xcd, 0x22, 0x31, 0x19, 0x08, 0xab, 0x42, 0x19, 0x15, 0xd9,
+	0xf6, 0x59, 0x6e, 0x09, 0xe0, 0xf4, 0x38, 0x18, 0x8d, 0x42, 0x31, 0x72, 0xa3, 0x18, 0xc6, 0x94,
+	0xc1, 0xee, 0x42, 0x99, 0x94, 0x63, 0x24, 0x8a, 0x5b, 0x5d, 0x49, 0x7b, 0xdb, 0xbf, 0xbe, 0x6c,
+	0x69, 0xbf, 0xbf, 0x6c, 0x69, 0x7f, 0xbe, 0x6c, 0x69, 0x3f, 0xfe, 0xd5, 0xda, 0xb8, 0x2c, 0xd3,
+	0xdf, 0x0c, 0xef, 0xfd, 0x1d, 0x00, 0x00, 0xff, 0xff, 0x70, 0xbe, 0xda, 0x79, 0x76, 0x10, 0x00,
+	0x00,
 }
 
 func (m *Row) Marshal() (dAtA []byte, err error) {
@@ -3639,6 +3649,11 @@ func (m *QueryRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if m.MaxMemory != 0 {
+		i = encodeVarintPublic(dAtA, i, uint64(m.MaxMemory))
+		i--
+		dAtA[i] = 0x50
+	}
 	if m.PreTranslated {
 		i--
 		if m.PreTranslated {
@@ -5262,6 +5277,9 @@ func (m *QueryRequest) Size() (n int) {
 	if m.PreTranslated {
 		n += 2
 	}
+	if m.MaxMemory != 0 {
+		n += 1 + sovPublic(uint64(m.MaxMemory))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -8577,6 +8595,25 @@ func (m *QueryRequest) Unmarshal(dAtA []byte) error {
 				}
 			}
 			m.PreTranslated = bool(v != 0)
+		case 10:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field MaxMemory", wireType)
+			}
+			m.MaxMemory = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPublic
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.MaxMemory |= int64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipPublic(dAtA[iNdEx:])
diff --git a/pb/public.proto b/pb/public.proto
index 1ecc2fabb..70935c647 100644
--- a/pb/public.proto
+++ b/pb/public.proto
@@ -122,6 +122,7 @@ message QueryRequest {
 	bool Remote = 5;
 	repeated Row EmbeddedData = 8;
 	bool PreTranslated = 9;
+	int64 MaxMemory = 10;
 }
 
 message QueryResponse {
diff --git a/server.go b/server.go
index b233ed935..233c61b42 100644
--- a/server.go
+++ b/server.go
@@ -449,8 +449,16 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	}
 	s.holderConfig.AntiEntropyInterval = s.antiEntropyInterval
 
+	memTotal, err := s.systemInfo.MemTotal()
+	if err != nil {
+		return nil, errors.Wrap(err, "mem total")
+	}
+
 	// set up executor after server opts have been processed
-	executorOpts := []executorOption{optExecutorInternalQueryClient(s.defaultClient)}
+	executorOpts := []executorOption{
+		optExecutorInternalQueryClient(s.defaultClient),
+		optExecutorMaxMemory(int64(float64(memTotal) * .50)),
+	}
 	if s.executorPoolSize > 0 {
 		executorOpts = append(executorOpts, optExecutorWorkerPoolSize(s.executorPoolSize))
 	}

From 60d534c505f56a7ac5e4349dda933d74cb3c59f1 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Mon, 2 Aug 2021 15:27:05 -0600
Subject: [PATCH 2/2] Limit translation memory & add max query memory config

---
 ctl/server.go    |  1 +
 executor.go      | 53 +++++++++++++++++++++++++++++++++++++-----------
 server.go        | 17 +++++++++++++++-
 server/config.go |  3 +++
 server/server.go |  3 ++-
 5 files changed, 63 insertions(+), 14 deletions(-)

diff --git a/ctl/server.go b/ctl/server.go
index b4f8b8029..2660a8b37 100644
--- a/ctl/server.go
+++ b/ctl/server.go
@@ -39,6 +39,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
 	flags.Uint64Var(&srv.Config.MaxFileCount, "max-file-count", srv.Config.MaxFileCount, "Soft limit on the maximum number of fragment files FeatureBase keeps open simultaneously.")
 	flags.DurationVar((*time.Duration)(&srv.Config.LongQueryTime), "long-query-time", time.Duration(srv.Config.LongQueryTime), "Duration that will trigger log and stat messages for slow queries. Zero to disable.")
 	flags.IntVar(&srv.Config.QueryHistoryLength, "query-history-length", srv.Config.QueryHistoryLength, "Number of queries to remember in history.")
+	flags.Int64Var(&srv.Config.MaxQueryMemory, "max-query-memory", srv.Config.MaxQueryMemory, "Maximum memory allowed per Extract() or SELECT query.")
 
 	// TLS
 	SetTLSConfig(flags, "", &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.CACertPath, &srv.Config.TLS.SkipVerify, &srv.Config.TLS.EnableClientVerification)
diff --git a/executor.go b/executor.go
index b0d1959d4..ec071f34f 100644
--- a/executor.go
+++ b/executor.go
@@ -205,6 +205,10 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
 	if opt == nil {
 		opt = &execOptions{}
 	}
+	// Default maximum memory, if not passed in.
+	if opt.MaxMemory == 0 {
+		opt.MaxMemory = e.maxMemory
+	}
 
 	if opt.Profile {
 		var prof tracing.ProfiledSpan
@@ -235,7 +239,7 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
 	// No need to translate a remote call.
 	if !opt.Remote {
 		// only translateResults if this local node is the final destination. only string/column keys.
-		if err := e.translateResults(ctx, index, idx, q.Calls, results); err != nil {
+		if err := e.translateResults(ctx, index, idx, q.Calls, results, opt.MaxMemory); err != nil {
 			if errors.Cause(err) == ErrTranslatingKeyNotFound {
 				// No error - return empty result
 				resp.Results = make([]interface{}, len(q.Calls))
@@ -3956,7 +3960,7 @@ func (e *executor) executeExternalLookup(ctx context.Context, qcx *Qcx, index st
 	}
 
 	qr := []interface{}{rawArg}
-	err = e.translateResults(ctx, index, idx, c.Children, qr)
+	err = e.translateResults(ctx, index, idx, c.Children, qr, e.maxMemory)
 	if err != nil {
 		return ExtractedTable{}, errors.Wrap(err, "translating query result")
 	}
@@ -4163,11 +4167,6 @@ func (e *executor) executeExternalLookup(ctx context.Context, qcx *Qcx, index st
 }
 
 func (e *executor) executeExtract(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (ExtractedIDMatrix, error) {
-	// Defualt maximum memory, if not passed in.
-	if opt.MaxMemory == 0 {
-		opt.MaxMemory = e.maxMemory
-	}
-
 	// Extract the column filter call.
 	if len(c.Children) < 1 {
 		return ExtractedIDMatrix{}, errors.New("missing column filter in Extract")
@@ -5852,12 +5851,13 @@ func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapRe
 // calcResultMemory recursively computes the total memory used by v.
 func calcResultMemory(v interface{}) (n int64) {
 	switch v := v.(type) {
-	case string:
 	case ExtractedIDColumn:
 		n += 8 // ColumnID
 		for _, row := range v.Rows {
 			n += 24 + int64(len(row)*8) // slice header + data
 		}
+		return n
+
 	case ExtractedIDMatrix:
 		n += 24 // slice size
 		for _, field := range v.Fields {
@@ -5868,8 +5868,34 @@ func calcResultMemory(v interface{}) (n int64) {
 		for _, col := range v.Columns {
 			n += calcResultMemory(col)
 		}
+		return n
+
+	case ExtractedTableColumn:
+		n += 8 + 16 + int64(len(v.Column.Key)) + 8 // KeyOrID
+		for _, row := range v.Rows {
+			n += 8 + calcResultMemory(row) // ptr + value size
+		}
+		return n
+
+	case string:
+		return 16 + int64(len(v))
+	case bool, int64, uint64:
+		return 8
+	case []string:
+		n += 24 // slice header
+		for i := range v {
+			n += 16 + int64(len(v[i]))
+		}
+		return n
+	case []uint64:
+		return 24 + int64(8*len(v)) // slice header + data size
+	case pql.Decimal:
+		return 16
+	case time.Time:
+		return 24
+	default:
+		return n
 	}
-	return n
 }
 
 type job struct {
@@ -6617,7 +6643,7 @@ func (e *executor) callZero(c *pql.Call) *pql.Call {
 	}
 }
 
-func (e *executor) translateResults(ctx context.Context, index string, idx *Index, calls []*pql.Call, results []interface{}) (err error) {
+func (e *executor) translateResults(ctx context.Context, index string, idx *Index, calls []*pql.Call, results []interface{}, memoryAvailable int64) (err error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "Executor.translateResults")
 	defer span.Finish()
 
@@ -6636,7 +6662,7 @@ func (e *executor) translateResults(ctx context.Context, index string, idx *Inde
 	}
 
 	for i := range results {
-		results[i], err = e.translateResult(ctx, index, idx, calls[i], results[i], idMap)
+		results[i], err = e.translateResult(ctx, index, idx, calls[i], results[i], idMap, &memoryAvailable)
 		if err != nil {
 			return err
 		}
@@ -6758,7 +6784,7 @@ func (e *executor) preTranslateMatrixSet(mat ExtractedIDMatrix, fieldIdx uint, f
 	return e.Cluster.translateFieldIDs(field, ids)
 }
 
-func (e *executor) translateResult(ctx context.Context, index string, idx *Index, call *pql.Call, result interface{}, idSet map[uint64]string) (_ interface{}, err error) {
+func (e *executor) translateResult(ctx context.Context, index string, idx *Index, call *pql.Call, result interface{}, idSet map[uint64]string, memoryAvailable *int64) (_ interface{}, err error) {
 	switch result := result.(type) {
 	case *Row:
 		rowIdx, rowField, strategy, err := e.howToTranslate(idx, result)
@@ -7198,6 +7224,9 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
 				Column: colTrans,
 				Rows:   data,
 			}
+			if *memoryAvailable -= calcResultMemory(cols[i]); *memoryAvailable < 0 {
+				return nil, fmt.Errorf("table exceeds available memory")
+			}
 		}
 
 		return ExtractedTable{
diff --git a/server.go b/server.go
index 233c61b42..b3b0b5128 100644
--- a/server.go
+++ b/server.go
@@ -91,6 +91,7 @@ type Server struct { // nolint: maligned
 	confirmDownSleep    time.Duration
 	confirmDownRetries  int
 	syncer              holderSyncer
+	maxQueryMemory      int64
 
 	translationSyncer      TranslationSyncer
 	resetTranslationSyncCh chan struct{}
@@ -368,6 +369,14 @@ func OptServerQueryHistoryLength(length int) ServerOption {
 	}
 }
 
+// OptServerMaxQueryMemory sets the memory used per Extract() and SELECT query.
+func OptServerMaxQueryMemory(v int64) ServerOption {
+	return func(s *Server) error {
+		s.maxQueryMemory = v
+		return nil
+	}
+}
+
 // OptServerDisCo is a functional option on Server
 // used to set the Distributed Consensus implementation.
 func OptServerDisCo(disCo disco.DisCo,
@@ -454,10 +463,16 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 		return nil, errors.Wrap(err, "mem total")
 	}
 
+	// Default memory to 20% of total.
+	maxQueryMemory := s.maxQueryMemory
+	if maxQueryMemory == 0 {
+		maxQueryMemory = int64(float64(memTotal) * .20)
+	}
+
 	// set up executor after server opts have been processed
 	executorOpts := []executorOption{
 		optExecutorInternalQueryClient(s.defaultClient),
-		optExecutorMaxMemory(int64(float64(memTotal) * .50)),
+		optExecutorMaxMemory(maxQueryMemory),
 	}
 	if s.executorPoolSize > 0 {
 		executorOpts = append(executorOpts, optExecutorWorkerPoolSize(s.executorPoolSize))
diff --git a/server/config.go b/server/config.go
index bb5b412c5..b96a5be1f 100644
--- a/server/config.go
+++ b/server/config.go
@@ -130,6 +130,9 @@ type Config struct {
 	// don't exhaust the goroutine limit.
 	ImportWorkerPoolSize int `toml:"-"`
 
+	// Limits the total amount of memory to be used by Extract() & SELECT queries.
+	MaxQueryMemory int64 `toml:"max-query-memory"`
+
 	Cluster struct {
 		ReplicaN int    `toml:"replicas"`
 		Name     string `toml:"name"`
diff --git a/server/server.go b/server/server.go
index 967dbd562..dd5488561 100644
--- a/server/server.go
+++ b/server/server.go
@@ -40,7 +40,6 @@ import (
 
 	"golang.org/x/sync/errgroup"
 
-	"github.com/pelletier/go-toml"
 	"github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/boltdb"
 	"github.com/molecula/featurebase/v2/encoding/proto"
@@ -56,6 +55,7 @@ import (
 	"github.com/molecula/featurebase/v2/statsd"
 	"github.com/molecula/featurebase/v2/syswrap"
 	"github.com/molecula/featurebase/v2/testhook"
+	"github.com/pelletier/go-toml"
 	"github.com/pkg/errors"
 )
 
@@ -493,6 +493,7 @@ func (m *Command) SetupServer() error {
 		pilosa.OptServerStorageConfig(m.Config.Storage),
 		pilosa.OptServerRowcacheOn(m.Config.RowcacheOn),
 		pilosa.OptServerRBFConfig(m.Config.RBFConfig),
+		pilosa.OptServerMaxQueryMemory(m.Config.MaxQueryMemory),
 		pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength),
 		discoOpt,
 	}