diff --git a/api.go b/api.go index cae91b20f..a9bd02f2d 100644 --- a/api.go +++ b/api.go @@ -38,6 +38,8 @@ type API struct { holder *Holder cluster *cluster server *Server + + Serializer Serializer } // APIOption is a functional option type for pilosa.API @@ -48,6 +50,7 @@ func OptAPIServer(s *Server) APIOption { a.server = s a.holder = s.holder a.cluster = s.cluster + a.Serializer = s.serializer return nil } } diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 10cc13dc4..0f2574f1b 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -2,6 +2,7 @@ package proto import ( "fmt" + "sort" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa" @@ -161,6 +162,23 @@ func (Serializer) Unmarshal(buf []byte, m pilosa.Message) error { } decodeNode(msg, mt) return nil + case *pilosa.QueryRequest: + msg := &internal.QueryRequest{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling QueryRequest") + } + decodeQueryRequest(msg, mt) + return nil + case *pilosa.QueryResponse: + msg := &internal.QueryResponse{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling QueryResponse") + } + decodeQueryResponse(msg, mt) + return nil + default: panic(fmt.Sprintf("unhandled pilosa.Message of type %T: %#v", mt, m)) } @@ -202,10 +220,62 @@ func encodeToProto(m pilosa.Message) proto.Message { return encodeNodeStatus(mt) case *pilosa.Node: return encodeNode(mt) + case *pilosa.QueryRequest: + return encodeQueryRequest(mt) + case *pilosa.QueryResponse: + return encodeQueryResponse(mt) } return nil } +func encodeQueryRequest(m *pilosa.QueryRequest) *internal.QueryRequest { + return &internal.QueryRequest{ + Query: m.Query, + Shards: m.Shards, + ColumnAttrs: m.ColumnAttrs, + Remote: m.Remote, + ExcludeRowAttrs: m.ExcludeRowAttrs, + ExcludeColumns: m.ExcludeColumns, + } +} + +func encodeQueryResponse(m *pilosa.QueryResponse) *internal.QueryResponse { + pb := &internal.QueryResponse{ + Results: make([]*internal.QueryResult, len(m.Results)), + ColumnAttrSets: EncodeColumnAttrSets(m.ColumnAttrSets), + } + + for i := range m.Results { + pb.Results[i] = &internal.QueryResult{} + + switch result := m.Results[i].(type) { + case *pilosa.Row: + pb.Results[i].Type = queryResultTypeRow + pb.Results[i].Row = EncodeRow(result) + case []pilosa.Pair: + pb.Results[i].Type = queryResultTypePairs + pb.Results[i].Pairs = EncodePairs(result) + case pilosa.ValCount: + pb.Results[i].Type = queryResultTypeValCount + pb.Results[i].ValCount = EncodeValCount(result) + case uint64: + pb.Results[i].Type = queryResultTypeUint64 + pb.Results[i].N = result + case bool: + pb.Results[i].Type = queryResultTypeBool + pb.Results[i].Changed = result + case nil: + pb.Results[i].Type = queryResultTypeNil + } + } + + if m.Err != nil { + pb.Err = m.Err.Error() + } + + return pb +} + func encodeResizeInstruction(m *pilosa.ResizeInstruction) *internal.ResizeInstruction { return &internal.ResizeInstruction{ JobID: m.JobID, @@ -602,3 +672,221 @@ func decodeNodeStatus(pb *internal.NodeStatus, m *pilosa.NodeStatus) { } func decodeRecalculateCaches(pb *internal.RecalculateCaches, m *pilosa.RecalculateCaches) {} + +func decodeQueryRequest(pb *internal.QueryRequest, m *pilosa.QueryRequest) { + m.Query = pb.Query + m.Shards = pb.Shards + m.ColumnAttrs = pb.ColumnAttrs + m.Remote = pb.Remote + m.ExcludeRowAttrs = pb.ExcludeRowAttrs + m.ExcludeColumns = pb.ExcludeColumns +} + +func decodeQueryResponse(pb *internal.QueryResponse, m *pilosa.QueryResponse) { + m.ColumnAttrSets = make([]*pilosa.ColumnAttrSet, len(pb.ColumnAttrSets)) + decodeColumnAttrSets(pb.ColumnAttrSets, m.ColumnAttrSets) + m.Err = errors.New(pb.Err) + m.Results = make([]interface{}, len(pb.Results)) + decodeQueryResults(pb.Results, m.Results) + +} + +func decodeColumnAttrSets(pb []*internal.ColumnAttrSet, m []*pilosa.ColumnAttrSet) { + for i := range pb { + decodeColumnAttrSet(pb[i], m[i]) + } +} + +func decodeColumnAttrSet(pb *internal.ColumnAttrSet, m *pilosa.ColumnAttrSet) { + m.ID = pb.ID + m.Key = pb.Key + m.Attrs = decodeAttrs(pb.Attrs) +} + +func decodeQueryResults(pb []*internal.QueryResult, m []interface{}) { + for i := range pb { + m[i] = decodeQueryResult(pb[i]) + } +} + +// QueryResult types. +const ( + queryResultTypeNil uint32 = iota + queryResultTypeRow + queryResultTypePairs + queryResultTypeValCount + queryResultTypeUint64 + queryResultTypeBool +) + +func decodeQueryResult(pb *internal.QueryResult) interface{} { + switch pb.Type { + case queryResultTypeRow: + return decodeRow(pb.Row) + case queryResultTypePairs: + return decodePairs(pb.Pairs) + case queryResultTypeValCount: + return decodeValCount(pb.ValCount) + case queryResultTypeUint64: + return pb.N + case queryResultTypeBool: + return pb.Changed + case queryResultTypeNil: + return nil + } + panic(fmt.Sprintf("unknown type: %d", pb.Type)) +} + +// DecodeRow converts r from its internal representation. +func decodeRow(pr *internal.Row) *pilosa.Row { + if pr == nil { + return nil + } + + r := pilosa.NewRow() + r.Attrs = decodeAttrs(pr.Attrs) + for _, v := range pr.Columns { + r.SetBit(v) + } + return r +} + +func decodeAttrs(pb []*internal.Attr) map[string]interface{} { + m := make(map[string]interface{}, len(pb)) + for i := range pb { + key, value := decodeAttr(pb[i]) + m[key] = value + } + return m +} + +const ( + attrTypeString = 1 + attrTypeInt = 2 + attrTypeBool = 3 + attrTypeFloat = 4 +) + +func decodeAttr(attr *internal.Attr) (key string, value interface{}) { + switch attr.Type { + case attrTypeString: + return attr.Key, attr.StringValue + case attrTypeInt: + return attr.Key, attr.IntValue + case attrTypeBool: + return attr.Key, attr.BoolValue + case attrTypeFloat: + return attr.Key, attr.FloatValue + default: + return attr.Key, nil + } +} + +func decodePairs(a []*internal.Pair) []pilosa.Pair { + other := make([]pilosa.Pair, len(a)) + for i := range a { + other[i] = decodePair(a[i]) + } + return other +} + +func decodePair(pb *internal.Pair) pilosa.Pair { + return pilosa.Pair{ + ID: pb.ID, + Key: pb.Key, + Count: pb.Count, + } +} + +func decodeValCount(pb *internal.ValCount) pilosa.ValCount { + return pilosa.ValCount{ + Val: pb.Val, + Count: pb.Count, + } +} + +func EncodeColumnAttrSets(a []*pilosa.ColumnAttrSet) []*internal.ColumnAttrSet { + other := make([]*internal.ColumnAttrSet, len(a)) + for i := range a { + other[i] = EncodeColumnAttrSet(a[i]) + } + return other +} + +func EncodeColumnAttrSet(set *pilosa.ColumnAttrSet) *internal.ColumnAttrSet { + return &internal.ColumnAttrSet{ + ID: set.ID, + Attrs: encodeAttrs(set.Attrs), + } +} + +func EncodeRow(r *pilosa.Row) *internal.Row { + if r == nil { + return nil + } + + return &internal.Row{ + Columns: r.Columns(), + Attrs: encodeAttrs(r.Attrs), + } +} + +func EncodePairs(a pilosa.Pairs) []*internal.Pair { + other := make([]*internal.Pair, len(a)) + for i := range a { + other[i] = encodePair(a[i]) + } + return other +} + +func encodePair(p pilosa.Pair) *internal.Pair { + return &internal.Pair{ + ID: p.ID, + Key: p.Key, + Count: p.Count, + } +} + +func EncodeValCount(vc pilosa.ValCount) *internal.ValCount { + return &internal.ValCount{ + Val: vc.Val, + Count: vc.Count, + } +} + +func encodeAttrs(m map[string]interface{}) []*internal.Attr { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + + a := make([]*internal.Attr, len(keys)) + for i := range keys { + a[i] = encodeAttr(keys[i], m[keys[i]]) + } + return a +} + +// encodeAttr converts a key/value pair into an Attr internal representation. +func encodeAttr(key string, value interface{}) *internal.Attr { + pb := &internal.Attr{Key: key} + switch value := value.(type) { + case string: + pb.Type = attrTypeString + pb.StringValue = value + case float64: + pb.Type = attrTypeFloat + pb.FloatValue = value + case uint64: + pb.Type = attrTypeInt + pb.IntValue = int64(value) + case int64: + pb.Type = attrTypeInt + pb.IntValue = value + case bool: + pb.Type = attrTypeBool + pb.BoolValue = value + } + return pb +} diff --git a/gossip/gossip.go b/gossip/gossip.go index 251851077..28b3ac690 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -28,7 +28,6 @@ import ( "github.com/hashicorp/memberlist" "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/encoding/proto" "github.com/pilosa/pilosa/toml" "github.com/pkg/errors" ) @@ -43,9 +42,8 @@ type GossipMemberSet struct { broadcasts *memberlist.TransmitLimitedQueue - papi *pilosa.API - serializer pilosa.Serializer - config *gossipConfig + papi *pilosa.API + config *gossipConfig Logger pilosa.Logger @@ -150,9 +148,8 @@ func WithLogger(logger *log.Logger) GossipMemberSetOption { func NewGossipMemberSet(cfg Config, api *pilosa.API, options ...GossipMemberSetOption) (*GossipMemberSet, error) { host := api.Node().URI.GetHost() g := &GossipMemberSet{ - papi: api, - serializer: proto.Serializer{}, - Logger: pilosa.NopLogger, + papi: api, + Logger: pilosa.NopLogger, } // options @@ -223,7 +220,7 @@ func NewGossipMemberSet(cfg Config, api *pilosa.API, options ...GossipMemberSetO // NodeMeta implementation of the memberlist.Delegate interface. func (g *GossipMemberSet) NodeMeta(limit int) []byte { - buf, err := g.serializer.Marshal(g.papi.Node()) + buf, err := g.papi.Serializer.Marshal(g.papi.Node()) if err != nil { g.Logger.Printf("marshal message error: %s", err) return []byte{} @@ -256,7 +253,7 @@ func (g *GossipMemberSet) LocalState(join bool) []byte { } // Marshal nodestate data to bytes. - buf, err := pilosa.MarshalInternalMessage(m, g.serializer) + buf, err := pilosa.MarshalInternalMessage(m, g.papi.Serializer) if err != nil { g.Logger.Printf("error marshalling nodestate data, err=%s", err) return []byte{} @@ -279,9 +276,8 @@ func (g *GossipMemberSet) MergeRemoteState(buf []byte, join bool) { // Care must be taken that events are processed in a timely manner from // the channel, since this delegate will block until an event can be sent. type gossipEventReceiver struct { - ch chan memberlist.NodeEvent - papi *pilosa.API - serializer pilosa.Serializer + ch chan memberlist.NodeEvent + papi *pilosa.API logger *log.Logger } @@ -289,10 +285,9 @@ type gossipEventReceiver struct { // newGossipEventReceiver returns a new instance of GossipEventReceiver. func newGossipEventReceiver(logger *log.Logger, papi *pilosa.API) *gossipEventReceiver { ger := &gossipEventReceiver{ - ch: make(chan memberlist.NodeEvent, 1), - logger: logger, - papi: papi, - serializer: proto.Serializer{}, + ch: make(chan memberlist.NodeEvent, 1), + logger: logger, + papi: papi, } go ger.listen() return ger @@ -327,7 +322,7 @@ func (g *gossipEventReceiver) listen() { // Get the node from the event.Node meta data. var n pilosa.Node - if err := g.serializer.Unmarshal(e.Node.Meta, &n); err != nil { + if err := g.papi.Serializer.Unmarshal(e.Node.Meta, &n); err != nil { panic("failed to unmarshal event node meta into node") } @@ -335,7 +330,7 @@ func (g *gossipEventReceiver) listen() { Event: nodeEventType, Node: &n, } - buf, err := pilosa.MarshalInternalMessage(ne, g.serializer) + buf, err := pilosa.MarshalInternalMessage(ne, g.papi.Serializer) if err != nil { panic(err) } diff --git a/http/handler.go b/http/handler.go index 9195f05e9..84832b1d4 100644 --- a/http/handler.go +++ b/http/handler.go @@ -815,13 +815,12 @@ func (h *Handler) readProtobufQueryRequest(r *http.Request) (*pilosa.QueryReques return nil, errors.Wrap(err, "reading") } - // Unmarshal into object. - var req internal.QueryRequest - if err := proto.Unmarshal(body, &req); err != nil { - return nil, errors.Wrap(err, "unmarshalling") + qreq := &pilosa.QueryRequest{} + err = h.API.Serializer.Unmarshal(body, qreq) + if err != nil { + return nil, errors.Wrap(err, "unmarshalling query request") } - - return decodeQueryRequest(&req), nil + return qreq, nil } // readURLQueryRequest parses query parameters from URL parameters from r. @@ -860,7 +859,7 @@ func (h *Handler) writeQueryResponse(w http.ResponseWriter, r *http.Request, res // writeProtobufQueryResponse writes the response from the executor to w as protobuf. func (h *Handler) writeProtobufQueryResponse(w http.ResponseWriter, resp *pilosa.QueryResponse) error { - if buf, err := proto.Marshal(encodeQueryResponse(resp)); err != nil { + if buf, err := h.API.Serializer.Marshal(resp); err != nil { return errors.Wrap(err, "marshalling") } else if _, err := w.Write(buf); err != nil { return errors.Wrap(err, "writing")