From ce815471fee78e7357cf14e9fe27cfe26daa7cde Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 1 Nov 2018 12:49:17 -0500 Subject: [PATCH 1/3] Drop now-superfluous methodNotAllowedHandler Long ago, the maintainers of gorilla/mux concluded that it was a "wontfix" to return StatusMethodNotAllowed instead of StatusNotFound for a method mismatch. Pilosa had a workaround for this for the most common case (GET requests to /index/{index}/query), and a TODO to address the other cases. While browsing the go-pilosa client code, I noticed that there is a test for roaring import support which relies on getting StatusMethodNotAllowed. But how can this work, if gorilla/mux doesn't do that? Answer: They started doing it in mid-2017, apparently: https://github.com/gorilla/mux/issues/271 Dropping this code changes the exact text of the message produced for that one case, but not the status code, and makes the behavior less confusing. --- http/handler.go | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/http/handler.go b/http/handler.go index 6007d72a8..1d5dee6f7 100644 --- a/http/handler.go +++ b/http/handler.go @@ -260,20 +260,10 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/internal/shards/max", handler.handleGetShardsMax).Methods("GET").Name("GetShardsMax") // TODO: deprecate, but it's being used by the client router.HandleFunc("/internal/translate/data", handler.handleGetTranslateData).Methods("GET").Name("GetTranslateData") - // TODO: Apply MethodNotAllowed statuses to all endpoints. - // Ideally this would be automatic, as described in this (wontfix) ticket: - // https://github.com/gorilla/mux/issues/6 - // For now we just do it for the most commonly used handler, /query - router.HandleFunc("/index/{index}/query", handler.methodNotAllowedHandler).Methods("GET") - router.Use(handler.queryArgValidator) return router } -func (h *Handler) methodNotAllowedHandler(w http.ResponseWriter, _ *http.Request) { - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) -} - // ServeHTTP handles an HTTP request. func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { defer func() { From 456c0929308c1ffe6dea65ea90d9d754edf1e745 Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 1 Nov 2018 13:56:48 -0500 Subject: [PATCH 2/3] enterprise/b: Ensure 64-bit range for btree keys. The cmp() function used to compare btree keys was using the trick of comparing unsigned values by coercing the result of subtraction to a signed type. This is fine as long as the range of differences never actually exceeds the limits of the signed type. For instance, with int64, as long as the magnitude of the difference is under 2^63 or so, it works reasonably well. Plain int, however, can be a 32-bit type, at which point the magnitude of difference needed to break it is only 2^31 or so. Subtraction and type conversion is enough cheaper than branches that this is worth preserving, but it's worth preserving by switching to an explicit int64 for the operation and return type. This will not actually affect performance except for people using the btree code on 32-bit machines, so it probably won't ever matter. Which may also be true of the potential wrong answers, but "might be slower" is a better risk than "might crash". --- enterprise/b/btree.go | 2 +- enterprise/b/containers_btree.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/enterprise/b/btree.go b/enterprise/b/btree.go index 7b9b34e76..c61f5c5b6 100644 --- a/enterprise/b/btree.go +++ b/enterprise/b/btree.go @@ -75,7 +75,7 @@ type ( // 0 if a == b // > 0 if a > b // - Cmp func(a, b uint64) int + Cmp func(a, b uint64) int64 d struct { // data page c int diff --git a/enterprise/b/containers_btree.go b/enterprise/b/containers_btree.go index 7487eea70..c2f95c0c7 100644 --- a/enterprise/b/containers_btree.go +++ b/enterprise/b/containers_btree.go @@ -23,8 +23,8 @@ import ( "github.com/pilosa/pilosa/roaring" ) -func cmp(a, b uint64) int { - return int(a - b) +func cmp(a, b uint64) int64 { + return int64(a - b) } type bTreeContainers struct { From 063caed8d1ef2eff6f40c757bec9533fe5a47a80 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Fri, 2 Nov 2018 19:17:07 -0500 Subject: [PATCH 3/3] forward imports to non-coordinator shards --- api.go | 171 ++++++++++++++++++++++++++--------- api_test.go | 232 ++++++++++++++++++++++++++++++++++++++++++++++++ http/client.go | 3 + http/handler.go | 12 ++- 4 files changed, 371 insertions(+), 47 deletions(-) create mode 100644 api_test.go diff --git a/api.go b/api.go index af206496b..dcedc3d92 100644 --- a/api.go +++ b/api.go @@ -690,7 +690,8 @@ func (api *API) FieldAttrDiff(_ context.Context, indexName string, fieldName str // ImportOptions holds the options for the API.Import method. type ImportOptions struct { - Clear bool + Clear bool + IgnoreKeyCheck bool } // ImportOption is a functional option type for API.Import. @@ -703,8 +704,15 @@ func OptImportOptionsClear(c bool) ImportOption { } } +func OptImportOptionsIgnoreKeyCheck(b bool) ImportOption { + return func(o *ImportOptions) error { + o.IgnoreKeyCheck = b + return nil + } +} + // Import bulk imports data into a particular index,field,shard. -func (api *API) Import(_ context.Context, req *ImportRequest, opts ...ImportOption) error { +func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOption) error { if err := api.validate(apiImport); err != nil { return errors.Wrap(err, "validating api method") } @@ -715,34 +723,71 @@ func (api *API) Import(_ context.Context, req *ImportRequest, opts ...ImportOpti return errors.Wrap(err, "setting up import options") } - index := api.holder.Index(req.Index) - if index == nil { - return newNotFoundError(ErrIndexNotFound) - } - - field, err := api.indexField(req.Index, req.Field, req.Shard) + index, field, err := api.indexField(req.Index, req.Field, req.Shard) if err != nil { - return errors.Wrap(err, "getting field") + return errors.Wrap(err, "getting index and field") } - // Translate row keys. - if field.keys() { - if len(req.RowIDs) != 0 { - return errors.New("row ids cannot be used because field uses string keys") + // Unless explicitly ignoring key validation (meaning keys have been + // translated to ids in a previous step at the coordinator node), then + // check to see if keys need translation. + if !options.IgnoreKeyCheck { + // Translate row keys. + if field.keys() { + if len(req.RowIDs) != 0 { + return errors.New("row ids cannot be used because field uses string keys") + } + if req.RowIDs, err = api.holder.translateFile.TranslateRowsToUint64(index.Name(), field.Name(), req.RowKeys); err != nil { + return errors.Wrap(err, "translating rows") + } } - if req.RowIDs, err = api.holder.translateFile.TranslateRowsToUint64(index.Name(), field.Name(), req.RowKeys); err != nil { - return errors.Wrap(err, "translating rows") + + // Translate column keys. + if index.Keys() { + if len(req.ColumnIDs) != 0 { + return errors.New("column ids cannot be used because index uses string keys") + } + if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil { + return errors.Wrap(err, "translating columns") + } + } + + // For translated data, map the columnIDs to shards. If + // this node does not own the shard, forward to the node that does. + if index.Keys() || field.keys() { + m := make(map[uint64][]Bit) + + for i, colID := range req.ColumnIDs { + shard := colID / ShardWidth + if _, ok := m[shard]; !ok { + m[shard] = make([]Bit, 0) + } + m[shard] = append(m[shard], Bit{ + RowID: req.RowIDs[i], + ColumnID: colID, + Timestamp: req.Timestamps[i], + }) + } + + // Signal to the receiving nodes to ignore checking for key translation. + opts = append(opts, OptImportOptionsIgnoreKeyCheck(true)) + + var eg errgroup.Group + for shard, bits := range m { + // TODO: if local node owns this shard we don't need to go through the client + shard := shard + bits := bits + eg.Go(func() error { + return api.server.defaultClient.Import(ctx, req.Index, req.Field, shard, bits, opts...) + }) + } + return eg.Wait() } } - // Translate column keys. - if index.Keys() { - if len(req.ColumnIDs) != 0 { - return errors.New("column ids cannot be used because index uses string keys") - } - if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil { - return errors.Wrap(err, "translating columns") - } + // Validate shard ownership. + if err := api.validateShardOwnership(req.Index, req.Shard); err != nil { + return errors.Wrap(err, "validating shard ownership") } // Convert timestamps to time.Time. @@ -772,7 +817,7 @@ func (api *API) Import(_ context.Context, req *ImportRequest, opts ...ImportOpti } // ImportValue bulk imports values into a particular field. -func (api *API) ImportValue(_ context.Context, req *ImportValueRequest, opts ...ImportOption) error { +func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts ...ImportOption) error { if err := api.validate(apiImportValue); err != nil { return errors.Wrap(err, "validating api method") } @@ -783,26 +828,60 @@ func (api *API) ImportValue(_ context.Context, req *ImportValueRequest, opts ... return errors.Wrap(err, "setting up import options") } - index := api.holder.Index(req.Index) - if index == nil { - return newNotFoundError(ErrIndexNotFound) - } - - field, err := api.indexField(req.Index, req.Field, req.Shard) + index, field, err := api.indexField(req.Index, req.Field, req.Shard) if err != nil { - return errors.Wrap(err, "getting field") + return errors.Wrap(err, "getting index and field") } - // Translate column keys. - if index.Keys() { - if len(req.ColumnIDs) != 0 { - return errors.New("column ids cannot be used because index uses string keys") - } - if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil { - return errors.Wrap(err, "translating columns") + // Unless explicitly ignoring key validation (meaning keys have been + // translate to ids in a previous step at the coordinator node), then + // check to see if keys need translation. + if !options.IgnoreKeyCheck { + // Translate column keys. + if index.Keys() { + if len(req.ColumnIDs) != 0 { + return errors.New("column ids cannot be used because index uses string keys") + } + if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil { + return errors.Wrap(err, "translating columns") + } + + // For translated data, map the columnIDs to shards. If + // this node does not own the shard, forward to the node that does. + m := make(map[uint64][]FieldValue) + + for i, colID := range req.ColumnIDs { + shard := colID / ShardWidth + if _, ok := m[shard]; !ok { + m[shard] = make([]FieldValue, 0) + } + m[shard] = append(m[shard], FieldValue{ + Value: req.Values[i], + ColumnID: colID, + }) + } + + // Signal to the receiving nodes to ignore checking for key translation. + opts = append(opts, OptImportOptionsIgnoreKeyCheck(true)) + + var eg errgroup.Group + for shard, vals := range m { + // TODO: if local node owns this shard we don't need to go through the client + shard := shard + vals := vals + eg.Go(func() error { + return api.server.defaultClient.ImportValue(ctx, req.Index, req.Field, shard, vals, opts...) + }) + } + return eg.Wait() } } + // Validate shard ownership. + if err := api.validateShardOwnership(req.Index, req.Shard); err != nil { + return errors.Wrap(err, "validating shard ownership") + } + // Import columnIDs into existence field. if !options.Clear { if err := importExistenceColumns(index, req.ColumnIDs); err != nil { @@ -863,28 +942,32 @@ func (api *API) LongQueryTime() time.Duration { return api.cluster.longQueryTime } -func (api *API) indexField(indexName string, fieldName string, shard uint64) (*Field, error) { +func (api *API) validateShardOwnership(indexName string, shard uint64) error { // Validate that this handler owns the shard. if !api.cluster.ownsShard(api.Node().ID, indexName, shard) { api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName) - return nil, ErrClusterDoesNotOwnShard + return ErrClusterDoesNotOwnShard } + return nil +} + +func (api *API) indexField(indexName string, fieldName string, shard uint64) (*Index, *Field, error) { + api.server.logger.Printf("importing: %v %v %v", indexName, fieldName, shard) // Find the Index. - api.server.logger.Printf("importing: %v %v %v", indexName, fieldName, shard) index := api.holder.Index(indexName) if index == nil { api.server.logger.Printf("fragment error: index=%s, field=%s, shard=%d, err=%s", indexName, fieldName, shard, ErrIndexNotFound.Error()) - return nil, newNotFoundError(ErrIndexNotFound) + return nil, nil, newNotFoundError(ErrIndexNotFound) } // Retrieve field. field := index.Field(fieldName) if field == nil { api.server.logger.Printf("field error: index=%s, field=%s, shard=%d, err=%s", indexName, fieldName, shard, ErrFieldNotFound.Error()) - return nil, ErrFieldNotFound + return nil, nil, ErrFieldNotFound } - return field, nil + return index, field, nil } // SetCoordinator makes a new Node the cluster coordinator. diff --git a/api_test.go b/api_test.go new file mode 100644 index 000000000..e569896f0 --- /dev/null +++ b/api_test.go @@ -0,0 +1,232 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pilosa_test + +import ( + "context" + "fmt" + "reflect" + "testing" + + "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" + "github.com/pilosa/pilosa/test" +) + +func TestAPI_Import(t *testing.T) { + c := test.MustRunCluster(t, 2, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node0"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + )}, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node1"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + )}, + ) + defer c.Close() + + m0 := c[0] + m1 := c[1] + + t.Run("RowIDColumnKey", func(t *testing.T) { + ctx := context.Background() + index := "rick" + field := "f" + + _, err := m0.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: true}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + _, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100)) + if err != nil { + t.Fatalf("creating field: %v", err) + } + + rowID := uint64(1) + timestamp := int64(0) + + // Generate some keyed records. + rowIDs := []uint64{} + colKeys := []string{} + timestamps := []int64{} + for i := 1; i <= 10; i++ { + rowIDs = append(rowIDs, rowID) + timestamps = append(timestamps, timestamp) + colKeys = append(colKeys, fmt.Sprintf("col%d", i)) + } + + // Import data with keys to the coordinator (node0) and verify that it gets + // translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher) + req := &pilosa.ImportRequest{ + Index: index, + Field: field, + Shard: 0, + RowIDs: rowIDs, + ColumnKeys: colKeys, + Timestamps: timestamps, + } + if err := m0.API.Import(ctx, req); err != nil { + t.Fatal(err) + } + + pql := fmt.Sprintf("Row(%s=%d)", field, rowID) + + // Query node0. + if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + t.Fatal(err) + } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { + t.Fatalf("unexpected column keys: %+v", keys) + } + + // Query node1. + if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + t.Fatal(err) + } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { + t.Fatalf("unexpected column keys: %+v", keys) + } + }) + + t.Run("RowKeyColumnID", func(t *testing.T) { + ctx := context.Background() + index := "rkci" + field := "f" + + _, err := m0.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: false}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + _, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100), pilosa.OptFieldKeys()) + if err != nil { + t.Fatalf("creating field: %v", err) + } + + rowKey := "rowkey" + + // Generate some keyed records. + rowKeys := []string{rowKey, rowKey, rowKey} + colIDs := []uint64{1, 2, pilosa.ShardWidth + 1} + timestamps := []int64{0, 0, 0} + + // Import data with keys to the coordinator (node0) and verify that it gets + // translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher) + req := &pilosa.ImportRequest{ + Index: index, + Field: field, + Shard: 0, + RowKeys: rowKeys, + ColumnIDs: colIDs, + Timestamps: timestamps, + } + if err := m0.API.Import(ctx, req); err != nil { + t.Fatal(err) + } + + pql := fmt.Sprintf("Row(%s=%s)", field, rowKey) + + // Query node0. + if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + t.Fatal(err) + } else if columns := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, colIDs) { + t.Fatalf("unexpected column ids: %+v", columns) + } + + // Query node1. + if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + t.Fatal(err) + } else if columns := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, colIDs) { + t.Fatalf("unexpected column ids: %+v", columns) + } + }) +} + +func TestAPI_ImportValue(t *testing.T) { + c := test.MustRunCluster(t, 2, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node0"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + )}, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node1"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + )}, + ) + defer c.Close() + + m0 := c[0] + m1 := c[1] + + t.Run("ValColumnKey", func(t *testing.T) { + ctx := context.Background() + index := "valck" + field := "f" + + _, err := m0.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: true}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + _, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeInt(0, 100)) + if err != nil { + t.Fatalf("creating field: %v", err) + } + + // Generate some keyed records. + values := []int64{} + colKeys := []string{} + for i := 1; i <= 10; i++ { + values = append(values, int64(i)) + colKeys = append(colKeys, fmt.Sprintf("col%d", i)) + } + + // Import data with keys to the coordinator (node0) and verify that it gets + // translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher) + req := &pilosa.ImportValueRequest{ + Index: index, + Field: field, + ColumnKeys: colKeys, + Values: values, + } + if err := m0.API.ImportValue(ctx, req); err != nil { + t.Fatal(err) + } + + pql := fmt.Sprintf("Range(%s>0)", field) + + // Query node0. + if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + t.Fatal(err) + } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { + t.Fatalf("unexpected column keys: %+v", keys) + } + + // Query node1. + if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + t.Fatal(err) + } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { + t.Fatalf("unexpected column keys: %+v", keys) + } + }) +} + +// offsetModHasher represents a simple, mod-based hashing offset by 1. +type offsetModHasher struct{} + +func (*offsetModHasher) Hash(key uint64, n int) int { + return int(key+1) % n +} diff --git a/http/client.go b/http/client.go index 06bb31479..75d3c75c7 100644 --- a/http/client.go +++ b/http/client.go @@ -437,6 +437,9 @@ func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, inde if opts.Clear { vals.Set("clear", "true") } + if opts.IgnoreKeyCheck { + vals.Set("ignoreKeyCheck", "true") + } url := fmt.Sprintf("%s?%s", u.String(), vals.Encode()) req, err := http.NewRequest("POST", url, bytes.NewReader(buf)) diff --git a/http/handler.go b/http/handler.go index 1d5dee6f7..81e31a82b 100644 --- a/http/handler.go +++ b/http/handler.go @@ -181,7 +181,7 @@ func (h *Handler) populateValidators() { h.validators["DeleteIndex"] = queryValidationSpecRequired() h.validators["PostField"] = queryValidationSpecRequired() h.validators["DeleteField"] = queryValidationSpecRequired() - h.validators["PostImport"] = queryValidationSpecRequired().Optional("clear") + h.validators["PostImport"] = queryValidationSpecRequired().Optional("clear", "ignoreKeyCheck") h.validators["PostImportRoaring"] = queryValidationSpecRequired().Optional("remote", "clear") h.validators["PostQuery"] = queryValidationSpecRequired().Optional("shards", "columnAttrs", "excludeRowAttrs", "excludeColumns") h.validators["GetInfo"] = queryValidationSpecRequired() @@ -987,6 +987,12 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { // If the clear flag is true, treat the import as clear bits. q := r.URL.Query() doClear := q.Get("clear") == "true" + doIgnoreKeyCheck := q.Get("ignoreKeyCheck") == "true" + + opts := []pilosa.ImportOption{ + pilosa.OptImportOptionsClear(doClear), + pilosa.OptImportOptionsIgnoreKeyCheck(doIgnoreKeyCheck), + } // Get index and field type to determine how to handle the // import data. @@ -1020,7 +1026,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { return } - if err := h.api.ImportValue(r.Context(), req, pilosa.OptImportOptionsClear(doClear)); err != nil { + if err := h.api.ImportValue(r.Context(), req, opts...); err != nil { switch errors.Cause(err) { case pilosa.ErrClusterDoesNotOwnShard: http.Error(w, err.Error(), http.StatusPreconditionFailed) @@ -1038,7 +1044,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { return } - if err := h.api.Import(r.Context(), req, pilosa.OptImportOptionsClear(doClear)); err != nil { + if err := h.api.Import(r.Context(), req, opts...); err != nil { switch errors.Cause(err) { case pilosa.ErrClusterDoesNotOwnShard: http.Error(w, err.Error(), http.StatusPreconditionFailed)