diff --git a/api.go b/api.go index 06c983c3e..715945339 100644 --- a/api.go +++ b/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. diff --git a/boltdb/translate.go b/boltdb/translate.go index 714470eb5..68941fe5c 100644 --- a/boltdb/translate.go +++ b/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)) diff --git a/client.go b/client.go index b1ddaab30..5143f2e6b 100644 --- a/client.go +++ b/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{} } diff --git a/cluster.go b/cluster.go index 4c0b8f624..6a139cad4 100644 --- a/cluster.go +++ b/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)) { diff --git a/executor.go b/executor.go index b878ab170..ae63759f8 100644 --- a/executor.go +++ b/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) } diff --git a/fragment.go b/fragment.go index f7304f7ea..d117b0459 100644 --- a/fragment.go +++ b/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 diff --git a/http/client.go b/http/client.go index 4b9e16f49..8fcff044d 100644 --- a/http/client.go +++ b/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() diff --git a/http/handler.go b/http/handler.go index 76ebd87dd..e5d23f7aa 100644 --- a/http/handler.go +++ b/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" { diff --git a/like.go b/like.go index d3fbebe54..4ff44d1f3 100644 --- a/like.go +++ b/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 } diff --git a/like_test.go b/like_test.go index 7e5268dea..88ad51ee4 100644 --- a/like_test.go +++ b/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) } } diff --git a/translate.go b/translate.go index 888ad06ef..5983bd0a1 100644 --- a/translate.go +++ b/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 {