execute like queries on the primary's key translation database

This works around an issue where unreplicated keys will not be matched everywhere.
This also avoids the cost of creating millions of bolt read transactions and allocating strings.

api.go

@@ -2040,6 +2040,15 @@ func (api *API) CreateFieldKeys(ctx context.Context, index, field string, keys .
 	return api.cluster.createFieldKeys(ctx, f, keys...)
 }
 
+// MatchField finds the IDs of all field keys matching a filter.
+func (api *API) MatchField(ctx context.Context, index, field string, like string) ([]uint64, error) {
+	f := api.holder.Field(index, field)
+	if f == nil {
+		return nil, newNotFoundError(ErrFieldNotFound, field)
+	}
+	return api.cluster.matchField(ctx, f, like)
+}
+
 // PrimaryReplicaNodeURL returns the URL of the cluster's primary replica.
 func (api *API) PrimaryReplicaNodeURL() url.URL {
 	// Create a snapshot of the cluster to use for node/partition calculations.

boltdb/translate.go

@@ -299,6 +299,35 @@ func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
 	return result, nil
 }
 
+// Match finds the IDs of all keys matching a filter.
+func (s *TranslateStore) Match(filter func([]byte) bool) ([]uint64, error) {
+	var matches []uint64
+	err := s.db.View(func(tx *bolt.Tx) error {
+		// This uses the id bucket instead of the key bucket so that matches are produced in sorted order.
+		idBucket := tx.Bucket(bucketIDs)
+		if idBucket == nil {
+			return errors.Errorf(errFmtTranslateBucketNotFound, bucketIDs)
+		}
+
+		return idBucket.ForEach(func(id, key []byte) error {
+			if bytes.Equal(key, emptyKey) {
+				key = nil
+			}
+
+			if filter(key) {
+				matches = append(matches, btou64(id))
+			}
+
+			return nil
+		})
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return matches, nil
+}
+
 func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64, error) {
 	ids := make([]uint64, 0, len(keys))
 

client.go

@@ -113,6 +113,8 @@ type InternalQueryClient interface {
 
 	CreateIndexKeysNode(ctx context.Context, uri *pnet.URI, index string, keys ...string) (map[string]uint64, error)
 	CreateFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, keys ...string) (map[string]uint64, error)
+
+	MatchFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, like string) ([]uint64, error)
 }
 
 type nopInternalQueryClient struct{}
@@ -149,6 +151,10 @@ func (n nopInternalQueryClient) CreateFieldKeysNode(ctx context.Context, uri *pn
 	return nil, nil
 }
 
+func (n nopInternalQueryClient) MatchFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, like string) ([]uint64, error) {
+	return nil, nil
+}
+
 func newNopInternalQueryClient() nopInternalQueryClient {
 	return nopInternalQueryClient{}
 }

cluster.go

@@ -1466,6 +1466,24 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str
 	return translations, nil
 }
 
+func (c *cluster) matchField(ctx context.Context, field *Field, like string) ([]uint64, error) {
+	// The primary is the only node that can match field keys, since it is the only node with all of the keys.
+	primary := c.primaryNode()
+	if primary == nil {
+		return nil, errors.Errorf("matching field(%s/%s) like %q - cannot find primary node", field.Index(), field.Name(), like)
+	}
+	if c.Node.ID == primary.ID {
+		// The local copy is the authoritative copy.
+		plan := planLike(like)
+		return field.TranslateStore().Match(func(key []byte) bool {
+			return matchLike(key, plan...)
+		})
+	}
+
+	// Forward the request to the primary.
+	return c.InternalClient.MatchFieldKeysNode(ctx, &primary.URI, field.Index(), field.Name(), like)
+}
+
 func (c *cluster) translateFieldIDs(field *Field, ids map[uint64]struct{}) (map[uint64]string, error) {
 	idList := make([]uint64, len(ids))
 	{

executor.go

@@ -2860,6 +2860,10 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
 		if err != nil {
 			return nil, errors.Wrap(err, "getting column")
 		}
+		_, hasLike, err := child.StringArg("like")
+		if err != nil {
+			return nil, errors.Wrap(err, "getting like")
+		}
 		fieldName, ok := child.Args["_field"].(string)
 		if !ok {
 			return nil, errors.Errorf("%s call must have field with valid (string) field name. Got %v of type %[2]T", child.Name, child.Args["_field"])
@@ -2873,7 +2877,7 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
 			bases[i] = f.bsiGroup(f.name).Base
 		}
 
-		if hasLimit || hasCol { // we need to perform this query cluster-wide ahead of executeGroupByShard
+		if hasLimit || hasCol || hasLike { // we need to perform this query cluster-wide ahead of executeGroupByShard
 			if idx, ok := child.Args["valueidx"].(int64); ok {
 				// The rows query was already completed on the initiating node.
 				childRows[i] = opt.EmbeddedData[idx].Columns()
@@ -3560,6 +3564,35 @@ func (e *executor) executeRows(ctx context.Context, qcx *Qcx, index string, c *p
 		return nil, err
 	}
 	results, _ := other.(RowIDs)
+
+	if !opt.Remote {
+		if like, hasLike, err := c.StringArg("like"); err != nil {
+			return nil, errors.Wrap(err, "getting like pattern")
+		} else if hasLike {
+			matches, err := e.Cluster.matchField(ctx, e.Holder.Field(index, fieldName), like)
+			if err != nil {
+				return nil, errors.Wrap(err, "matching like pattern")
+			}
+
+			i, j, k := 0, 0, 0
+			for i < len(results) && j < len(matches) {
+				x, y := results[i], matches[j]
+				switch {
+				case x < y:
+					i++
+				case y < x:
+					j++
+				default:
+					results[k] = x
+					i++
+					j++
+					k++
+				}
+			}
+			results = results[:k]
+		}
+	}
+
 	return results, nil
 }
 
@@ -3675,14 +3708,6 @@ func (e *executor) executeRowsShard(ctx context.Context, qcx *Qcx, index string,
 		limit = int(lim)
 	}
 
-	var likeErr chan error
-	if like, hasLike, err := c.StringArg("like"); err != nil {
-		return nil, errors.Wrap(err, "getting like pattern")
-	} else if hasLike {
-		likeErr = make(chan error, 1)
-		filters = append(filters, NewBitmapLikeFilter(like, f.TranslateStore()))
-	}
-
 	tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard})
 	if err != nil {
 		return nil, err
@@ -3701,11 +3726,6 @@ func (e *executor) executeRowsShard(ctx context.Context, qcx *Qcx, index string,
 		if err != nil {
 			return nil, err
 		}
-		select {
-		case err = <-likeErr:
-			return nil, err
-		default:
-		}
 		rowIDs = rowIDs.merge(viewRows, limit)
 	}
 

fragment.go

@@ -3148,47 +3148,6 @@ func (f *fragment) minRowID(tx Tx) (uint64, bool, error) {
 	return min / ShardWidth, ok, err
 }
 
-// BitmapLikeFilter is a roaring.BitmapFilter which handles Like expressions.
-type BitmapLikeFilter struct {
-	roaring.BitmapRowFilterBase
-	plan       []filterStep
-	translator TranslateStore
-}
-
-var _ roaring.BitmapFilter = &BitmapLikeFilter{}
-
-func (b *BitmapLikeFilter) ConsiderKey(key roaring.FilterKey, n int32) roaring.FilterResult {
-	res, done := b.DetermineByKey(key)
-	if done {
-		return res
-	}
-	if n == 0 {
-		return key.RejectOne()
-	}
-	row := key.Row()
-	keyStr, err := b.translator.TranslateID(row)
-	if err != nil {
-		return b.SetResult(key, key.Fail(errors.Wrap(err, "translating key for row")))
-	}
-	if matchLike(keyStr, b.plan...) {
-		return b.SetResult(key, key.MatchRow())
-	}
-	return b.SetResult(key, key.RejectRow())
-}
-
-func (b *BitmapLikeFilter) ConsiderData(key roaring.FilterKey, data *roaring.Container) roaring.FilterResult {
-	b.FilterResult.Err = errors.New("like filter should not need to look at data")
-	return b.FilterResult
-}
-
-func NewBitmapLikeFilter(like string, translator TranslateStore) *BitmapLikeFilter {
-	return &BitmapLikeFilter{
-		BitmapRowFilterBase: *roaring.NewBitmapRowFilterBase(nil),
-		plan:                planLike(like),
-		translator:          translator,
-	}
-}
-
 // rows returns all rows starting from 'start'. Filters will be applied in
 // order. All filters must return true to include the row. Once a row is
 // included, further containers in that row will be skipped. So, for a row to be

http/client.go

@@ -27,6 +27,7 @@ import (
 	"path"
 	"sort"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/pilosa/pilosa/v2"
@@ -1466,6 +1467,49 @@ func (c *InternalClient) CreateFieldKeysNode(ctx context.Context, uri *pnet.URI,
 	return transMap, nil
 }
 
+func (c *InternalClient) MatchFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, like string) (matches []uint64, err error) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.MatchFieldKeysNode")
+	defer span.Finish()
+
+	// Create HTTP request.
+	u := uriPathToURL(uri, fmt.Sprintf("/internal/translate/field/%s/%s/keys/like", index, field))
+	req, err := http.NewRequest("POST", u.String(), strings.NewReader(like))
+	if err != nil {
+		return nil, errors.Wrap(err, "creating request")
+	}
+
+	// Apply headers.
+	req.Header.Set("Content-Length", strconv.Itoa(len(like)))
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
+
+	// Send the request.
+	resp, err := c.executeRequest(req.WithContext(ctx))
+	if err != nil {
+		return nil, errors.Wrap(err, "executing request")
+	}
+	defer func() {
+		cerr := resp.Body.Close()
+		if cerr != nil && err == nil {
+			err = errors.Wrap(cerr, "closing response body")
+		}
+	}()
+
+	// Read the response body.
+	result, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, errors.Wrap(err, "reading response")
+	}
+
+	// Decode the translations.
+	err = json.Unmarshal(result, &matches)
+	if err != nil {
+		return nil, errors.Wrap(err, "json decoding")
+	}
+
+	return matches, nil
+}
+
 func (c *InternalClient) Transactions(ctx context.Context) (map[string]*pilosa.Transaction, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Transactions")
 	defer span.Finish()

http/handler.go

@@ -363,7 +363,7 @@ func (h *Handler) collectStats(next http.Handler) http.Handler {
 
 // latticeRoutes lists the frontend routes that do not directly correspond to
 // backend routes, and require special handling.
-var latticeRoutes = []string{"/tables", "/query", "/querybuilder"}  // TODO somehow pull this from some metadata in the lattice directory
+var latticeRoutes = []string{"/tables", "/query", "/querybuilder"} // TODO somehow pull this from some metadata in the lattice directory
 
 // newRouter creates a new mux http router.
 func newRouter(handler *Handler) http.Handler {
@@ -435,6 +435,7 @@ func newRouter(handler *Handler) http.Handler {
 	router.HandleFunc("/internal/translate/field/{index}/{field}", handler.handlePostTranslateFieldDB).Methods("POST").Name("PostTranslateFieldDB")
 	router.HandleFunc("/internal/translate/field/{index}/{field}/keys/find", handler.handleFindFieldKeys).Methods("POST").Name("FindFieldKeys")
 	router.HandleFunc("/internal/translate/field/{index}/{field}/keys/create", handler.handleCreateFieldKeys).Methods("POST").Name("CreateFieldKeys")
+	router.HandleFunc("/internal/translate/field/{index}/{field}/keys/like", handler.handleMatchField).Methods("POST").Name("MatchFieldKeys")
 
 	router.HandleFunc("/internal/idalloc/reserve", handler.handleReserveIDs).Methods("POST").Name("ReserveIDs")
 	router.HandleFunc("/internal/idalloc/commit", handler.handleCommitIDs).Methods("POST").Name("CommitIDs")
@@ -2806,6 +2807,44 @@ func (h *Handler) handleCreateFieldKeys(w http.ResponseWriter, r *http.Request)
 	}
 }
 
+func (h *Handler) handleMatchField(w http.ResponseWriter, r *http.Request) {
+	// Verify output type.
+	if !validHeaderAcceptJSON(r.Header) {
+		http.Error(w, "Not acceptable", http.StatusNotAcceptable)
+		return
+	}
+
+	indexName, ok := mux.Vars(r)["index"]
+	if !ok {
+		http.Error(w, "index name is required", http.StatusBadRequest)
+		return
+	}
+
+	fieldName, ok := mux.Vars(r)["field"]
+	if !ok {
+		http.Error(w, "field name is required", http.StatusBadRequest)
+		return
+	}
+
+	bd, err := readBody(r)
+	if err != nil {
+		http.Error(w, "failed to read body", http.StatusBadRequest)
+		return
+	}
+
+	matches, err := h.api.MatchField(r.Context(), indexName, fieldName, string(bd))
+	if err != nil {
+		http.Error(w, "failed to match pattern", http.StatusInternalServerError)
+		return
+	}
+
+	err = json.NewEncoder(w).Encode(matches)
+	if err != nil {
+		http.Error(w, "encoding result", http.StatusBadRequest)
+		return
+	}
+}
+
 func (h *Handler) handleReserveIDs(w http.ResponseWriter, r *http.Request) {
 	// Verify input and output types
 	if r.Header.Get("Content-Type") != "application/json" {

like.go

@@ -15,6 +15,7 @@
 package pilosa
 
 import (
+	"bytes"
 	"strings"
 	"unicode/utf8"
 )
@@ -66,7 +67,7 @@ type filterStep struct {
 	kind filterStepKind
 
 	// str is the substring for a prefix/skipthrough/suffix step.
-	str string
+	str []byte
 
 	// n is the number of underscores in the step (if relevant).
 	n int
@@ -95,7 +96,7 @@ func planLike(like string) []filterStep {
 			// Generate a step to skip through the next token.
 			step = filterStep{
 				kind: filterStepSkipThrough,
-				str:  tokens[i+1],
+				str:  []byte(tokens[i+1]),
 				n:    underscores,
 			}
 			merged = true
@@ -115,7 +116,7 @@ func planLike(like string) []filterStep {
 			// Generate a step to process an exact match of the beginning of a string.
 			step = filterStep{
 				kind: filterStepPrefix,
-				str:  t,
+				str:  []byte(t),
 			}
 		}
 		steps = append(steps, step)
@@ -130,12 +131,12 @@ func planLike(like string) []filterStep {
 }
 
 // matchLike matches a string using a like plan.
-func matchLike(key string, like ...filterStep) bool {
+func matchLike(key []byte, like ...filterStep) bool {
 	for i, step := range like {
 		switch step.kind {
 		case filterStepPrefix:
 			// Match a prefix.
-			if !strings.HasPrefix(key, step.str) {
+			if !bytes.HasPrefix(key, step.str) {
 				return false
 			}
 			key = key[len(step.str):]
@@ -143,7 +144,7 @@ func matchLike(key string, like ...filterStep) bool {
 			// Skip some placeholders.
 			n := step.n
 			for j := 0; j < n; j++ {
-				_, len := utf8.DecodeRuneInString(key)
+				_, len := utf8.DecodeRune(key)
 				if len == 0 {
 					return false
 				}
@@ -155,7 +156,7 @@ func matchLike(key string, like ...filterStep) bool {
 			// Skip through placeholders.
 			var skipped int
 			for skipped < step.n {
-				j := strings.Index(key, step.str)
+				j := bytes.Index(key, step.str)
 				switch j {
 				case -1:
 					// There are no more matches.
@@ -164,7 +165,7 @@ func matchLike(key string, like ...filterStep) bool {
 					// Skip a single rune to ensure forward progress.
 					// This is somewhat inefficient since we have to search the string again next time.
 					// This will hopefully not have to be used very frequently.
-					_, len := utf8.DecodeRuneInString(key)
+					_, len := utf8.DecodeRune(key)
 					key = key[len:]
 					skipped += len
 				default:
@@ -182,7 +183,7 @@ func matchLike(key string, like ...filterStep) bool {
 			remaining := like[i+1:]
 			for {
 				// Find the next substring match.
-				j := strings.Index(key, step.str)
+				j := bytes.Index(key, step.str)
 				switch {
 				case j == -1:
 					// There are no more matches.
@@ -199,12 +200,12 @@ func matchLike(key string, like ...filterStep) bool {
 				}
 
 				// Skip the first rune of the substring so we do not rescan this substring match.
-				_, len := utf8.DecodeRuneInString(key)
+				_, len := utf8.DecodeRune(key)
 				key = key[len:]
 			}
 		case filterStepSuffix:
 			// Match a suffix.
-			if !strings.HasSuffix(key, step.str) {
+			if !bytes.HasSuffix(key, step.str) {
 				// Suffix not present.
 				return false
 			}
@@ -222,18 +223,13 @@ func matchLike(key string, like ...filterStep) bool {
 				return false
 			}
 
-			// Count the runes.
-			j := -1
-			for j = range key {
-			}
-
 			// Check if the string is long enough.
-			return j+1 >= step.n
+			return utf8.RuneCount(key) >= step.n
 		default:
 			panic("invalid step")
 		}
 	}
 
 	// If there is any unmatched data left, this is not a match.
-	return key == ""
+	return len(key) == 0
 }

like_test.go

@@ -20,7 +20,6 @@ import (
 )
 
 func TestPlanLike(t *testing.T) {
-
 	cases := []struct {
 		name            string
 		like            string
@@ -40,7 +39,7 @@ func TestPlanLike(t *testing.T) {
 			plan: []filterStep{
 				{
 					kind: filterStepPrefix,
-					str:  "x",
+					str:  []byte("x"),
 				},
 			},
 			match:    []string{"x"},
@@ -63,7 +62,7 @@ func TestPlanLike(t *testing.T) {
 			plan: []filterStep{
 				{
 					kind: filterStepPrefix,
-					str:  "x",
+					str:  []byte("x"),
 				},
 				{
 					kind: filterStepMinLength,
@@ -79,7 +78,7 @@ func TestPlanLike(t *testing.T) {
 			plan: []filterStep{
 				{
 					kind: filterStepSuffix,
-					str:  "x",
+					str:  []byte("x"),
 				},
 			},
 			match:    []string{"x", "xx", "ax"},
@@ -91,11 +90,11 @@ func TestPlanLike(t *testing.T) {
 			plan: []filterStep{
 				{
 					kind: filterStepPrefix,
-					str:  "x",
+					str:  []byte("x"),
 				},
 				{
 					kind: filterStepSuffix,
-					str:  "y",
+					str:  []byte("y"),
 				},
 			},
 			match:    []string{"xy", "xzy", "xyzzy"},
@@ -107,15 +106,15 @@ func TestPlanLike(t *testing.T) {
 			plan: []filterStep{
 				{
 					kind: filterStepPrefix,
-					str:  "x",
+					str:  []byte("x"),
 				},
 				{
 					kind: filterStepSkipThrough,
-					str:  "y",
+					str:  []byte("y"),
 				},
 				{
 					kind: filterStepSuffix,
-					str:  "z",
+					str:  []byte("z"),
 				},
 			},
 			match:    []string{"xyz", "xzyzz", "x.y.z", "x.y.y..z"},
@@ -127,7 +126,7 @@ func TestPlanLike(t *testing.T) {
 			plan: []filterStep{
 				{
 					kind: filterStepPrefix,
-					str:  "a",
+					str:  []byte("a"),
 				},
 				{
 					kind: filterStepSkipN,
@@ -135,16 +134,16 @@ func TestPlanLike(t *testing.T) {
 				},
 				{
 					kind: filterStepPrefix,
-					str:  "b",
+					str:  []byte("b"),
 				},
 				{
 					kind: filterStepSkipThrough,
-					str:  "c",
+					str:  []byte("c"),
 					n:    2,
 				},
 				{
 					kind: filterStepSuffix,
-					str:  "d",
+					str:  []byte("d"),
 					n:    3,
 				},
 			},
@@ -181,7 +180,7 @@ func TestPlanLike(t *testing.T) {
 			plan: []filterStep{
 				{
 					kind: filterStepPrefix,
-					str:  "x",
+					str:  []byte("x"),
 				},
 				{
 					kind: filterStepSkipN,
@@ -189,7 +188,7 @@ func TestPlanLike(t *testing.T) {
 				},
 				{
 					kind: filterStepPrefix,
-					str:  "y",
+					str:  []byte("y"),
 				},
 			},
 			match:    []string{"x.y", "xay", "x y", "x⊕y"},
@@ -232,12 +231,12 @@ func TestPlanLike(t *testing.T) {
 				t.Parallel()
 
 				for _, m := range c.match {
-					if !matchLike(m, c.plan...) {
+					if !matchLike([]byte(m), c.plan...) {
 						t.Errorf("key %q was not matched", m)
 					}
 				}
 				for _, nm := range c.nonmatch {
-					if matchLike(nm, c.plan...) {
+					if matchLike([]byte(nm), c.plan...) {
 						t.Errorf("key %q was matched", nm)
 					}
 				}

translate.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"sort"
 	"sync"
 
 	"github.com/pilosa/pilosa/v2/topology"
@@ -81,6 +82,9 @@ type TranslateStore interface { // TODO: refactor this interface; readonly shoul
 	// If the translator is read-only, this will return an error.
 	CreateKeys(keys ...string) (map[string]uint64, error)
 
+	// Match finds IDs of strings matching the filter.
+	Match(filter func([]byte) bool) ([]uint64, error)
+
 	// Converts an integer ID to its associated string key.
 	TranslateID(id uint64) (string, error)
 	TranslateIDs(id []uint64) ([]string, error)
@@ -477,6 +481,23 @@ func (s *InMemTranslateStore) CreateKeys(keys ...string) (map[string]uint64, err
 	return result, nil
 }
 
+func (s *InMemTranslateStore) Match(filter func([]byte) bool) ([]uint64, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	var matches []uint64
+	for key, id := range s.idsByKey {
+		if filter([]byte(key)) {
+			matches = append(matches, id)
+		}
+	}
+	sort.Slice(matches, func(i, j int) bool {
+		return matches[i] < matches[j]
+	})
+
+	return matches, nil
+}
+
 func (s *InMemTranslateStore) translateKey(key string, writable bool) (_ uint64, err error) {
 	id := s.idsByKey[key]
 	if id != 0 {