ImportRoaring, add remote arg, fix data copy bug

also, handle err properly in client method instead of discarding.

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)
 		}

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 {

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
 }
 

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) {

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()