scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Import roaring enpoint accepts a list of views

Browse source
This commit is contained in:
Yuce Tekol 2018-11-12 18:43:59 +03:00
parent 8c99bd7a15
commit 84c04900e1
No known key found for this signature in database
GPG key ID: CB59E46D2FB90573
12 changed files with 1067 additions and 2660 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
api.go
View file

@ -267,17 +267,11 @@ func setUpImportOptions(opts ...ImportOption) (*ImportOptions, error) {
// (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, remote bool, data []byte, opts ...ImportOption) (err error) {
func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, shard uint64, remote bool, req *ImportRoaringRequest) (err error) {
if err = api.validate(apiField); err != nil {
return errors.Wrap(err, "validating api method")
}
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
return errors.Wrap(err, "setting up import options")
}
nodes := api.cluster.shardNodes(indexName, shard)
var eg errgroup.Group
@ -294,18 +288,26 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string,
for _, node := range nodes {
node := node
if node.ID == api.server.nodeID {
// must make a copy of data to operate on locally. field.importRoaring changes data
d2 := make([]byte, len(data))
copy(d2, data)
eg.Go(func() error {
return field.importRoaring(d2, shard, options.Clear)
var err error
for _, view := range req.Views {
// must make a copy of data to operate on locally.
// field.importRoaring changes data
data := make([]byte, len(view.Data))
copy(data, view.Data)
err = field.importRoaring(data, shard, view.Name, req.Clear)
if err != nil {
return err
}
}
return err
})
go func(node *Node) {
}(node)
} else if !remote { // if remote == true we don't forward to other nodes
// forward it on
eg.Go(func() error {
return api.server.defaultClient.ImportRoaring(ctx, &node.URI, indexName, fieldName, shard, true, data, opts...)
return api.server.defaultClient.ImportRoaring(ctx, &node.URI, indexName, fieldName, shard, true, req)
})
}
}

4
client.go
View file

@ -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, uri *URI, index, field string, shard uint64, remote bool, data []byte, opts ...ImportOption) error
ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) 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, opts ...ImportOption) error {
return nil
}
func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, data []byte, opts ...ImportOption) error {
func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error {
return nil
}
func (n nopInternalClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error {

42
encoding/proto/proto.go
View file

@ -217,6 +217,14 @@ func (Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
}
decodeImportValueRequest(msg, mt)
return nil
case *pilosa.ImportRoaringRequest:
msg := &internal.ImportRoaringRequest{}
err := proto.Unmarshal(buf, msg)
if err != nil {
return errors.Wrap(err, "unmarshaling ImportRoaringRequest")
}
decodeImportRoaringRequest(msg, mt)
return nil
case *pilosa.ImportResponse:
msg := &internal.ImportResponse{}
err := proto.Unmarshal(buf, msg)
@ -292,6 +300,8 @@ func encodeToProto(m pilosa.Message) proto.Message {
return encodeImportRequest(mt)
case *pilosa.ImportValueRequest:
return encodeImportValueRequest(mt)
case *pilosa.ImportRoaringRequest:
return encodeImportRoaringRequest(mt)
case *pilosa.ImportResponse:
return encodeImportResponse(mt)
case *pilosa.BlockDataRequest:
@ -348,6 +358,24 @@ func encodeImportValueRequest(m *pilosa.ImportValueRequest) *internal.ImportValu
}
}
func encodeImportRoaringRequestView(m *pilosa.ImportRoaringRequestView) *internal.ImportRoaringRequestView {
return &internal.ImportRoaringRequestView{
Name: m.Name,
Data: m.Data,
}
}
func encodeImportRoaringRequest(m *pilosa.ImportRoaringRequest) *internal.ImportRoaringRequest {
views := make([]*internal.ImportRoaringRequestView, len(m.Views))
for i, view := range m.Views {
views[i] = encodeImportRoaringRequestView(&view)
}
return &internal.ImportRoaringRequest{
Clear: m.Clear,
Views: views,
}
}
func encodeQueryRequest(m *pilosa.QueryRequest) *internal.QueryRequest {
return &internal.QueryRequest{
Query: m.Query,
@ -914,6 +942,20 @@ func decodeImportValueRequest(pb *internal.ImportValueRequest, m *pilosa.ImportV
m.Values = pb.Values
}
func decodeImportRoaringRequestView(pb *internal.ImportRoaringRequestView, m *pilosa.ImportRoaringRequestView) {
m.Name = pb.Name
m.Data = pb.Data
}
func decodeImportRoaringRequest(pb *internal.ImportRoaringRequest, m *pilosa.ImportRoaringRequest) {
views := make([]pilosa.ImportRoaringRequestView, len(pb.Views))
for i, view := range pb.Views {
decodeImportRoaringRequestView(view, &views[i])
}
m.Clear = pb.Clear
m.Views = views
}
func decodeImportResponse(pb *internal.ImportResponse, m *pilosa.ImportResponse) {
m.Err = pb.Err
}

7
field.go
View file

@ -1182,9 +1182,10 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO
return nil
}
func (f *Field) importRoaring(data []byte, shard uint64, clear bool) error {
viewName := viewStandard
func (f *Field) importRoaring(data []byte, shard uint64, viewName string, clear bool) error {
if viewName == "" {
viewName = viewStandard
}
view, err := f.createViewIfNotExists(viewName)
if err != nil {
return errors.Wrap(err, "creating view")

10
handler.go
View file

@ -96,6 +96,16 @@ type ImportRequest struct {
Timestamps []int64
}
type ImportRoaringRequestView struct {
Name string
Data []byte
}
type ImportRoaringRequest struct {
Clear bool
Views []ImportRoaringRequestView
}
type ImportResponse struct {
Err string
}

31
http/client.go
View file

@ -550,7 +550,7 @@ 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, uri *pilosa.URI, index, field string, shard uint64, remote bool, data []byte, opts ...pilosa.ImportOption) error {
func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, remote bool, req *pilosa.ImportRoaringRequest) error {
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
@ -560,32 +560,27 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, ind
uri = c.defaultURI
}
// Set up import options.
options := &pilosa.ImportOptions{}
for _, opt := range opts {
err := opt(options)
if err != nil {
return errors.Wrap(err, "applying option")
}
}
vals := url.Values{}
vals.Set("remote", strconv.FormatBool(remote))
if options.Clear {
vals.Set("clear", "true")
}
url := fmt.Sprintf("%s/index/%s/field/%s/import-roaring/%d?%s", uri, index, field, shard, vals.Encode())
// Marshal data to protobuf.
data, err := c.serializer.Marshal(req)
if err != nil {
return errors.Wrap(err, "marshal import request")
}
// Generate HTTP request.
req, err := http.NewRequest("POST", url, bytes.NewBuffer(data))
httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(data))
if err != nil {
return errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Type", "application/x-binary")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
httpReq.Header.Set("Content-Type", "application/x-protobuf")
httpReq.Header.Set("Accept", "application/x-protobuf")
httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
resp, err := c.executeRequest(httpReq.WithContext(ctx))
if err != nil {
return err
}
@ -595,7 +590,7 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, ind
rbody := &pilosa.ImportResponse{}
dec.Decode(rbody)
if rbody.Err != "" {
return errors.Errorf("importing roaring: %v", rbody.Err)
return errors.Wrap(errors.New(rbody.Err), "importing roaring")
}
return nil
}

32
http/client_test.go
View file

@ -408,8 +408,9 @@ func TestClient_ImportRoaring(t *testing.T) {
// Send import request.
host := cluster[0].URL()
c := MustNewClient(host, http.GetHTTPClient(nil))
roaringData, _ := hex.DecodeString("3B3001000100000900010000000100010009000100") // [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringData); err != nil {
// [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
roaringReq := makeImportRoaringRequest(false, "3B3001000100000900010000000100010009000100")
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringReq); err != nil {
t.Fatal(err)
}
@ -432,8 +433,9 @@ func TestClient_ImportRoaring(t *testing.T) {
}
// Ensure that sending a roaring import with the clear flag works as expected.
roaringDataClear, _ := hex.DecodeString("3A30000001000000010001001000000003000400") // [65539, 65540]
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringDataClear, pilosa.OptImportOptionsClear(true)); err != nil {
// [65539, 65540]
roaringReq = makeImportRoaringRequest(true, "3A30000001000000010001001000000003000400")
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringReq); err != nil {
t.Fatal(err)
}
@ -454,8 +456,9 @@ func TestClient_ImportRoaring(t *testing.T) {
}
// Ensure that sending a roaring import with the clear flag works as expected.
roaringDataClear, _ = hex.DecodeString("3A300000020000000000010001000100180000001C0000000400060001000300") // [4, 6, 65537, 65539]
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringDataClear, pilosa.OptImportOptionsClear(true)); err != nil {
// [4, 6, 65537, 65539]
roaringReq = makeImportRoaringRequest(true, "3A300000020000000000010001000100180000001C0000000400060001000300")
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringReq); err != nil {
t.Fatal(err)
}
@ -476,8 +479,9 @@ func TestClient_ImportRoaring(t *testing.T) {
}
// Ensure that sending a roaring import with the clear flag works as expected.
roaringDataClear, _ = hex.DecodeString("3B3001000100000900010000000100010009000100") // [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringDataClear, pilosa.OptImportOptionsClear(true)); err != nil {
// [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
roaringReq = makeImportRoaringRequest(true, "3B3001000100000900010000000100010009000100")
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringReq); err != nil {
t.Fatal(err)
}
@ -981,3 +985,15 @@ func MustNewClient(host string, h *gohttp.Client) *Client {
}
return &Client{InternalClient: c}
}
func makeImportRoaringRequest(clear bool, viewData string) *pilosa.ImportRoaringRequest {
roaringData, _ := hex.DecodeString(viewData)
view := pilosa.ImportRoaringRequestView{
Name: "",
Data: roaringData,
}
return &pilosa.ImportRoaringRequest{
Clear: clear,
Views: []pilosa.ImportRoaringRequestView{view},
}
}

25
http/handler.go
View file

@ -24,9 +24,8 @@ import (
"io/ioutil"
"net"
"net/http"
"net/url"
// Imported for its side-effect of registering pprof endpoints with the server.
_ "net/http/pprof"
"net/url" // Imported for its side-effect of registering pprof endpoints with the server.
"reflect"
"runtime/debug"
"strconv"
@ -36,7 +35,6 @@ import (
"github.com/gorilla/handlers"
"github.com/gorilla/mux"
"github.com/pilosa/pilosa"
"github.com/pkg/errors"
)
@ -1496,10 +1494,18 @@ func GetHTTPClient(t *tls.Config) *http.Client {
// handlPostRoaringImport
func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Content-Type") != "application/x-binary" {
// Verify that request is only communicating over protobufs.
if r.Header.Get("Content-Type") != "application/x-protobuf" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
} else if r.Header.Get("Accept") != "application/x-protobuf" {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName := mux.Vars(r)["index"]
fieldName := mux.Vars(r)["field"]
q := r.URL.Query()
remoteStr := q.Get("remote")
var remote bool
@ -1507,9 +1513,6 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
remote = true
}
// If the clear flag is true, treat the import as clear bits.
doClear := q.Get("clear") == "true"
// Read entire body.
body, err := ioutil.ReadAll(r.Body)
if err != nil {
@ -1517,6 +1520,12 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
return
}
req := &pilosa.ImportRoaringRequest{}
if err := h.api.Serializer.Unmarshal(body, req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
urlVars := mux.Vars(r)
shard, err := strconv.ParseUint(urlVars["shard"], 10, 64)
if err != nil {
@ -1526,7 +1535,7 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
resp := &pilosa.ImportResponse{}
// TODO give meaningful stats for import
err = h.api.ImportRoaring(r.Context(), urlVars["index"], urlVars["field"], shard, remote, body, pilosa.OptImportOptionsClear(doClear))
err = h.api.ImportRoaring(r.Context(), indexName, fieldName, shard, remote, req)
if err != nil {
resp.Err = err.Error()
if _, ok := err.(pilosa.BadRequestError); ok {

2051
internal/private.pb.go
View file

File diff suppressed because it is too large Load diff

1480
internal/public.pb.go
View file

File diff suppressed because it is too large Load diff

10
internal/public.proto
View file

@ -105,3 +105,13 @@ message ImportValueRequest {
repeated string ColumnKeys = 7;
repeated int64 Values = 6;
}
message ImportRoaringRequestView {
string Name = 1;
bytes Data = 2;
}
message ImportRoaringRequest {
bool Clear = 1;
repeated ImportRoaringRequestView views = 2;
}

9
server/handler_test.go
View file

@ -22,14 +22,13 @@ import (
"fmt"
"io"
"io/ioutil"
gohttp "net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"time"
gohttp "net/http"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/server"
@ -91,9 +90,9 @@ func TestHandler_Endpoints(t *testing.T) {
t.Run("ImportRoaring", func(t *testing.T) {
w := httptest.NewRecorder()
roaringData, _ := hex.DecodeString("3B3001000100000900010000000100010009000100")
req := test.MustNewHTTPRequest("POST", "/index/i0/field/f1/import-roaring/0", bytes.NewBuffer(roaringData))
req.Header.Set("Content-Type", "application/x-binary")
h.ServeHTTP(w, req)
httpReq := test.MustNewHTTPRequest("POST", "/index/i0/field/f1/import-roaring/0", bytes.NewBuffer(roaringData))
httpReq.Header.Set("Content-Type", "application/x-binary")
h.ServeHTTP(w, httpReq)
resp, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i0", Query: "TopN(f1)"})
if err != nil {
t.Fatalf("querying: %v", err)