scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-07 11:27:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #249 from seebs/roaringrow

Browse source 
use roaring row support internally
This commit is contained in:
seebs 2020-05-15 17:00:17 -05:00 • committed by GitHub
commit c6d61391d2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 376 additions and 348 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

650
encoding/proto/proto.go
View file

File diff suppressed because it is too large Load diff

7
http/client.go
View file

@ -291,6 +291,7 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index s
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
@ -495,6 +496,7 @@ func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, inde
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
@ -682,6 +684,7 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, ind
}
httpReq.Header.Set("Content-Type", "application/x-protobuf")
httpReq.Header.Set("Accept", "application/x-protobuf")
httpReq.Header.Set("X-Pilosa-Row", "roaring")
httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
@ -731,6 +734,7 @@ func (c *InternalClient) ImportColumnAttrs(ctx context.Context, uri *pilosa.URI,
}
httpReq.Header.Set("Content-Type", "application/x-protobuf")
httpReq.Header.Set("Accept", "application/x-protobuf")
httpReq.Header.Set("X-Pilosa-Row", "roaring")
httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
@ -1006,6 +1010,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index,
req.Header.Set("Content-Type", "application/protobuf")
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Accept", "application/protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
resp, err := c.executeRequest(req.WithContext(ctx))
@ -1163,6 +1168,7 @@ func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pilosa.URI,
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
@ -1213,6 +1219,7 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI,
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.

63
http/handler.go
View file

@ -31,11 +31,13 @@ import (
"runtime/debug"
"strconv"
"strings"
"sync"
"time"
"github.com/gorilla/handlers"
"github.com/gorilla/mux"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/encoding/proto"
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/tracing"
@ -121,8 +123,18 @@ func OptHandlerCloseTimeout(d time.Duration) handlerOption {
}
}
var makeImportOk sync.Once
var importOk []byte
// NewHandler returns a new instance of Handler with a default logger.
func NewHandler(opts ...handlerOption) (*Handler, error) {
makeImportOk.Do(func() {
var err error
importOk, err = proto.DefaultSerializer.Marshal(&pilosa.ImportResponse{Err: ""})
if err != nil {
panic(fmt.Sprintf("trying to cache import-OK response: %v", err))
}
})
handler := &Handler{
logger: logger.NopLogger,
closeTimeout: time.Second * 30,
@ -472,6 +484,17 @@ func validHeaderAcceptJSON(header http.Header) bool {
return true
}
// headerAcceptRoaringRow tells us that the request should accept roaring
// rows in response.
func headerAcceptRoaringRow(header http.Header) bool {
for _, v := range header["X-Pilosa-Row"] {
if v == "roaring" {
return true
}
}
return false
}
// handleGetSchema handles GET /schema requests.
func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {
@ -1223,7 +1246,7 @@ func (h *Handler) readProtobufQueryRequest(r *http.Request) (*pilosa.QueryReques
}
qreq := &pilosa.QueryRequest{}
err = h.api.Serializer.Unmarshal(body, qreq)
err = proto.DefaultSerializer.Unmarshal(body, qreq)
if err != nil {
return nil, errors.Wrap(err, "unmarshalling query request")
}
@ -1271,15 +1294,19 @@ func (h *Handler) readURLQueryRequest(r *http.Request) (*pilosa.QueryRequest, er
func (h *Handler) writeQueryResponse(w http.ResponseWriter, r *http.Request, resp *pilosa.QueryResponse) error {
if !validHeaderAcceptJSON(r.Header) {
w.Header().Set("Content-Type", "application/protobuf")
return h.writeProtobufQueryResponse(w, resp)
return h.writeProtobufQueryResponse(w, resp, headerAcceptRoaringRow(r.Header))
}
w.Header().Set("Content-Type", "application/json")
return h.writeJSONQueryResponse(w, resp)
}
// writeProtobufQueryResponse writes the response from the executor to w as protobuf.
func (h *Handler) writeProtobufQueryResponse(w io.Writer, resp *pilosa.QueryResponse) error {
if buf, err := h.api.Serializer.Marshal(resp); err != nil {
func (h *Handler) writeProtobufQueryResponse(w io.Writer, resp *pilosa.QueryResponse, writeRoaring bool) error {
serializer := proto.DefaultSerializer
if writeRoaring {
serializer = proto.RoaringSerializer
}
if buf, err := serializer.Marshal(resp); err != nil {
return errors.Wrap(err, "marshalling")
} else if _, err := w.Write(buf); err != nil {
return errors.Wrap(err, "writing")
@ -1342,7 +1369,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
// Field type: Int
// Marshal into request object.
req := &pilosa.ImportValueRequest{}
if err := h.api.Serializer.Unmarshal(body, req); err != nil {
if err := proto.DefaultSerializer.Unmarshal(body, req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
@ -1360,7 +1387,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
// Field type: set, time, mutex
// Marshal into request object.
req := &pilosa.ImportRequest{}
if err := h.api.Serializer.Unmarshal(body, req); err != nil {
if err := proto.DefaultSerializer.Unmarshal(body, req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
@ -1376,15 +1403,8 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
}
}
// Marshal response object.
buf, e := h.api.Serializer.Marshal(&pilosa.ImportResponse{Err: ""})
if e != nil {
http.Error(w, "marshal import response", http.StatusInternalServerError)
return
}
// Write response.
_, err = w.Write(buf)
_, err = w.Write(importOk)
if err != nil {
h.logger.Printf("writing import response: %v", err)
}
@ -1885,7 +1905,7 @@ func (h *Handler) handlePostImportColumnAttrs(w http.ResponseWriter, r *http.Req
}
req := &pilosa.ImportColumnAttrsRequest{}
if err := h.api.Serializer.Unmarshal(body, req); err != nil {
if err := proto.DefaultSerializer.Unmarshal(body, req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
@ -1895,15 +1915,8 @@ func (h *Handler) handlePostImportColumnAttrs(w http.ResponseWriter, r *http.Req
return
}
// Marshal response object.
buf, e := h.api.Serializer.Marshal(&pilosa.ImportResponse{Err: ""})
if e != nil {
http.Error(w, "marshal import-column-attrs response", http.StatusInternalServerError)
return
}
// Write response.
_, err = w.Write(buf)
_, err = w.Write(importOk)
if err != nil {
h.logger.Printf("writing import-column-attrs response: %v", err)
}
@ -1943,7 +1956,7 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
req := &pilosa.ImportRoaringRequest{}
span, _ = tracing.StartSpanFromContext(ctx, "Unmarshal")
err = h.api.Serializer.Unmarshal(body, req)
err = proto.DefaultSerializer.Unmarshal(body, req)
span.Finish()
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
@ -1969,7 +1982,7 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
}
// Marshal response object.
buf, err := h.api.Serializer.Marshal(resp)
buf, err := proto.DefaultSerializer.Marshal(resp)
if err != nil {
http.Error(w, fmt.Sprintf("marshal import-roaring response: %v", err), http.StatusInternalServerError)
return

4
snapshotqueue.go
View file

@ -192,7 +192,9 @@ func (sq *prioritySnapshotQueue) Stop() {
sq.urgent = nil
close(sq.background)
sq.background = nil
sq.logger.Printf("snapshot queue: enqueued %d, skipped %d\n", sq.stats.enqueued, sq.stats.skipped)
if sq.stats.skipped > 0 || sq.stats.enqueued > 1 {
sq.logger.Printf("snapshot queue: enqueued %d, skipped %d\n", sq.stats.enqueued, sq.stats.skipped)
}
}
// Enqueue tries to add a fragment to the queue, if the fragment is not already