From f943c17e0b0dfe8ca1ea4c163db8fdca2d937f31 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 13 Sep 2018 15:03:31 -0500 Subject: [PATCH] ImportRoaring, add remote arg, fix data copy bug also, handle err properly in client method instead of discarding. --- api.go | 11 ++++++---- client.go | 4 ++-- http/client.go | 20 +++++++++++------ http/client_test.go | 53 +++++++++++++++++++++++++++++++++++++++++++++ http/handler.go | 8 ++++++- 5 files changed, 82 insertions(+), 14 deletions(-) diff --git a/api.go b/api.go index ffddc0702..a369f3d4b 100644 --- a/api.go +++ b/api.go @@ -312,7 +312,7 @@ func (api *API) Field(_ context.Context, indexName, fieldName string) (*Field, e // (shard*ShardWidth)+(i%ShardWidth). That is to say that "data" represents all // of the rows in this shard of this field concatenated together in one long // bitmap. -func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, shard uint64, data []byte) (err error) { +func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, shard uint64, remote bool, data []byte) (err error) { if err = api.validate(apiField); err != nil { return errors.Wrap(err, "validating api method") } @@ -326,15 +326,18 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, return newNotFoundError(ErrFieldNotFound) } wg.Add(1) + // must make a copy of data to operate on locally. field.importRoaring changes data + d2 := make([]byte, len(data)) + copy(d2, data) go func(node *Node) { - err = field.importRoaring(data, shard) + err = field.importRoaring(d2, shard) wg.Done() }(node) - } else { + } else if !remote { // if remote == true we don't forward to other nodes wg.Add(1) // forward it on go func(node *Node) { - err = api.server.defaultClient.ImportRoaring(ctx, node, indexName, fieldName, shard, data) + err = api.server.defaultClient.ImportRoaring(ctx, &node.URI, indexName, fieldName, shard, true, data) wg.Done() }(node) } diff --git a/client.go b/client.go index 124b2c6c5..a2ca0bf96 100644 --- a/client.go +++ b/client.go @@ -53,7 +53,7 @@ type InternalClient interface { RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) SendMessage(ctx context.Context, uri *URI, msg []byte) error RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error) - ImportRoaring(ctx context.Context, node *Node, index, field string, shard uint64, data []byte) error + ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, data []byte) error } //=============== @@ -109,7 +109,7 @@ func (n nopInternalClient) Import(ctx context.Context, index, field string, shar func (n nopInternalClient) ImportK(ctx context.Context, index, field string, bits []Bit) error { return nil } -func (n nopInternalClient) ImportRoaring(ctx context.Context, node *Node, index, field string, shard uint64, data []byte) error { +func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, data []byte) error { return nil } func (n nopInternalClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error { diff --git a/http/client.go b/http/client.go index 1ab8bd04b..922ac6234 100644 --- a/http/client.go +++ b/http/client.go @@ -527,23 +527,24 @@ func (c *InternalClient) marshalImportValuePayload(index, field string, shard ui // ImportRoaring does fast import of raw bits in roaring format (pilosa or // official format, see API.ImportRoaring). -func (c *InternalClient) ImportRoaring(ctx context.Context, node *pilosa.Node, index, field string, shard uint64, data []byte) error { +func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, remote bool, data []byte) error { if index == "" { return pilosa.ErrIndexRequired } else if field == "" { return pilosa.ErrFieldRequired } + if uri == nil { + uri = c.defaultURI + } - endpoint := fmt.Sprintf("/index/%s/field/%s/import-roaring/%d", index, field, shard) + url := fmt.Sprintf("%s/index/%s/field/%s/import-roaring/%d?remote=%v", uri, index, field, shard, remote) - // Create URL. - u := nodePathToURL(node, endpoint) // Generate HTTP request. - req, err := http.NewRequest("POST", u.String(), bytes.NewBuffer(data)) + req, err := http.NewRequest("POST", url, bytes.NewBuffer(data)) if err != nil { return errors.Wrap(err, "creating request") } - req.Header.Set("Accept", "application/x-binary") + req.Header.Set("Content-Type", "application/x-binary") req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) // Execute request against the host. @@ -557,8 +558,13 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, node *pilosa.Node, i if resp.StatusCode != http.StatusOK { return fmt.Errorf("invalid status: %d", resp.StatusCode) } - io.Copy(ioutil.Discard, resp.Body) + dec := json.NewDecoder(resp.Body) + rbody := &pilosa.ImportResponse{} + dec.Decode(rbody) + if rbody.Err != "" { + return errors.Errorf("importing roaring: %v", rbody.Err) + } return nil } diff --git a/http/client_test.go b/http/client_test.go index 131161d3c..fdb677265 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -18,6 +18,7 @@ import ( "bufio" "bytes" "context" + "encoding/hex" "fmt" gohttp "net/http" "reflect" @@ -363,6 +364,58 @@ func TestClient_Import(t *testing.T) { } } +// Ensure client can bulk import data. +func TestClient_ImportRoaring(t *testing.T) { + cluster := test.MustNewCluster(t, 2) + for _, c := range cluster { + c.Config.Cluster.ReplicaN = 2 + } + err := cluster.Start() + if err != nil { + t.Fatalf("starting cluster: %v", err) + } + defer cluster.Close() + + _, err = cluster[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + _, err = cluster[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 100)) + if err != nil { + t.Fatalf("creating field: %v", err) + } + _, err = cluster[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "Set(0, f=1)"}) + if err != nil { + t.Fatalf("querying: %v", err) + } + + // Send import request. + host := cluster[0].URL() + c := MustNewClient(host, http.GetHTTPClient(nil)) + roaringData, _ := hex.DecodeString("3B3001000100000900010000000100010009000100") + if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringData); err != nil { + t.Fatal(err) + } + + hldr := test.Holder{Holder: cluster[0].Server.Holder()} + // Verify data. + if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) { + t.Fatalf("unexpected columns: %+v", a) + } + if a := hldr.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) { + t.Fatalf("unexpected columns: %+v", a) + } + + hldr2 := test.Holder{Holder: cluster[1].Server.Holder()} + // Verify data. + if a := hldr2.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) { + t.Fatalf("unexpected columns: %+v", a) + } + if a := hldr2.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) { + t.Fatalf("unexpected columns: %+v", a) + } +} + // Ensure client can bulk import data. func TestClient_ImportKeys(t *testing.T) { t.Run("SingleNode", func(t *testing.T) { diff --git a/http/handler.go b/http/handler.go index ac325de5c..bc108568a 100644 --- a/http/handler.go +++ b/http/handler.go @@ -1436,6 +1436,12 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType) return } + q := r.URL.Query() + remoteStr := q.Get("remote") + var remote bool + if remoteStr == "true" { + remote = true + } // Read entire body. body, err := ioutil.ReadAll(r.Body) @@ -1452,7 +1458,7 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request } // TODO give meaningful stats for import - err = h.api.ImportRoaring(r.Context(), urlVars["index"], urlVars["field"], shard, body) + err = h.api.ImportRoaring(r.Context(), urlVars["index"], urlVars["field"], shard, remote, body) resp := &pilosa.ImportResponse{} if err != nil { resp.Err = err.Error()