From 063caed8d1ef2eff6f40c757bec9533fe5a47a80 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Fri, 2 Nov 2018 19:17:07 -0500 Subject: [PATCH] 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)