Merge pull request #1620 from niaow/fix-like

[CORE-577] Execute like queries on the primary's key translation database

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 {