diff --git a/api.go b/api.go index bea6e1e76..2c25dcd66 100644 --- a/api.go +++ b/api.go @@ -277,6 +277,19 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str return field, nil } +// Field retrieves the named field. +func (api *API) Field(ctx context.Context, indexName, fieldName string) (*Field, error) { + if err := api.validate(apiField); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + + field := api.holder.Field(indexName, fieldName) + if field == nil { + return nil, NewNotFoundError(ErrFieldNotFound) + } + return field, nil +} + // DeleteField removes the named field from the named index. If the index is not // found, an error is returned. If the field is not found, it is ignored and no // action is taken. @@ -866,6 +879,7 @@ const ( apiExportCSV apiFragmentBlockData apiFragmentBlocks + apiField apiFieldAttrDiff //apiHosts // not implemented apiImport @@ -909,6 +923,7 @@ var methodsNormal = map[apiMethod]struct{}{ apiExportCSV: struct{}{}, apiFragmentBlockData: struct{}{}, apiFragmentBlocks: struct{}{}, + apiField: struct{}{}, apiFieldAttrDiff: struct{}{}, apiImport: struct{}{}, apiImportValue: struct{}{}, diff --git a/apimethod_string.go b/apimethod_string.go index 7004e7258..881b79472 100644 --- a/apimethod_string.go +++ b/apimethod_string.go @@ -4,9 +4,9 @@ package pilosa import "strconv" -const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiMarshalFragmentapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiUnmarshalFragmentapiViews" +const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiMarshalFragmentapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiUnmarshalFragmentapiViews" -var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 73, 86, 98, 118, 135, 151, 160, 174, 182, 198, 216, 224, 244, 257, 271, 288, 301, 321, 329} +var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 73, 86, 98, 118, 135, 143, 159, 168, 182, 190, 206, 224, 232, 252, 265, 279, 296, 309, 329, 337} func (i apiMethod) String() string { if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) { diff --git a/http/client.go b/http/client.go index c56e50721..db017fb6c 100644 --- a/http/client.go +++ b/http/client.go @@ -293,7 +293,7 @@ func (c *InternalClient) Import(ctx context.Context, index, field string, shard // Import to each node. for _, node := range nodes { - if err := c.importNode(ctx, node, buf); err != nil { + if err := c.importNode(ctx, node, index, field, buf); err != nil { return fmt.Errorf("import node: host=%s, err=%s", node.URI, err) } } @@ -319,7 +319,7 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, colum } // Import to node. - if err := c.importNode(ctx, node, buf); err != nil { + if err := c.importNode(ctx, node, index, field, buf); err != nil { return fmt.Errorf("import node: host=%s, err=%s", node.URI, err) } @@ -386,9 +386,10 @@ func marshalImportPayloadK(index, field string, bits []pilosa.Bit) ([]byte, erro } // importNode sends a pre-marshaled import request to a node. -func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, buf []byte) error { +func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, index, field string, buf []byte) error { // Create URL & HTTP request. - u := nodePathToURL(node, "/import") + path := fmt.Sprintf("/index/%s/field/%s/import", index, field) + u := nodePathToURL(node, path) req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf)) if err != nil { return errors.Wrap(err, "creating request") @@ -444,7 +445,7 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s // Import to each node. for _, node := range nodes { - if err := c.importValueNode(ctx, node, buf); err != nil { + if err := c.importValueNode(ctx, node, index, field, buf); err != nil { return fmt.Errorf("import node: host=%s, err=%s", node.URI, err) } } @@ -473,9 +474,10 @@ func marshalImportValuePayload(index, field string, shard uint64, vals []pilosa. } // importValueNode sends a pre-marshaled import request to a node. -func (c *InternalClient) importValueNode(ctx context.Context, node *pilosa.Node, buf []byte) error { +func (c *InternalClient) importValueNode(ctx context.Context, node *pilosa.Node, index, field string, buf []byte) error { // Create URL & HTTP request. - u := nodePathToURL(node, "/import-value") + path := fmt.Sprintf("/index/%s/field/%s/import", index, field) + u := nodePathToURL(node, path) req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf)) if err != nil { return errors.Wrap(err, "creating request") diff --git a/http/handler.go b/http/handler.go index a24beb34a..9d6b9e080 100644 --- a/http/handler.go +++ b/http/handler.go @@ -195,8 +195,6 @@ func NewRouter(handler *Handler) *mux.Router { router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux).Methods("GET") router.Handle("/debug/vars", expvar.Handler()).Methods("GET") router.HandleFunc("/export", handler.handleGetExport).Methods("GET").Name("GetExport") - router.HandleFunc("/import", handler.handlePostImport).Methods("POST") - router.HandleFunc("/import-value", handler.handlePostImportValue).Methods("POST") router.HandleFunc("/index", handler.handleGetIndexes).Methods("GET") router.HandleFunc("/index/{index}", handler.handleGetIndex).Methods("GET") router.HandleFunc("/index/{index}", handler.handlePostIndex).Methods("POST") @@ -204,6 +202,7 @@ func NewRouter(handler *Handler) *mux.Router { //router.HandleFunc("/index/{index}/field", handler.handleGetFields).Methods("GET") // Not implemented. router.HandleFunc("/index/{index}/field/{field}", handler.handlePostField).Methods("POST") router.HandleFunc("/index/{index}/field/{field}", handler.handleDeleteField).Methods("DELETE") + router.HandleFunc("/index/{index}/field/{field}/import", handler.handlePostImport).Methods("POST") router.HandleFunc("/index/{index}/query", handler.handlePostQuery).Methods("POST").Name("PostQuery") router.HandleFunc("/info", handler.handleGetInfo).Methods("GET") router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST") @@ -886,60 +885,24 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { http.Error(w, "Not acceptable", http.StatusNotAcceptable) return } + indexName := mux.Vars(r)["index"] + fieldName := mux.Vars(r)["field"] - // Read entire body. - body, err := ioutil.ReadAll(r.Body) + // Get index and field type to determine how to handle the + // import data. + field, err := h.API.Field(r.Context(), indexName, fieldName) if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - // Marshal into request object. - var req internal.ImportRequest - if err := proto.Unmarshal(body, &req); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - if err := h.API.Import(r.Context(), req); err != nil { switch errors.Cause(err) { case pilosa.ErrIndexNotFound: fallthrough case pilosa.ErrFieldNotFound: http.Error(w, err.Error(), http.StatusNotFound) - case pilosa.ErrClusterDoesNotOwnShard: - http.Error(w, err.Error(), http.StatusPreconditionFailed) default: http.Error(w, err.Error(), http.StatusInternalServerError) } return } - // Marshal response object. - buf, e := proto.Marshal(&internal.ImportResponse{Err: errorString(err)}) - if e != nil { - http.Error(w, fmt.Sprintf("marshal import response: %s", err), http.StatusInternalServerError) - return - } - - // Write response. - if err != nil { - w.WriteHeader(http.StatusInternalServerError) - } - w.Write(buf) -} - -// handlePostImportValue handles /import-value requests. -func (h *Handler) handlePostImportValue(w http.ResponseWriter, r *http.Request) { - // 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 - } - // Read entire body. body, err := ioutil.ReadAll(r.Body) if err != nil { @@ -947,38 +910,53 @@ func (h *Handler) handlePostImportValue(w http.ResponseWriter, r *http.Request) return } - // Marshal into request object. - var req internal.ImportValueRequest - if err := proto.Unmarshal(body, &req); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - if err = h.API.ImportValue(r.Context(), req); err != nil { - switch errors.Cause(err) { - case pilosa.ErrIndexNotFound: - fallthrough - case pilosa.ErrFieldNotFound: - http.Error(w, err.Error(), http.StatusNotFound) - case pilosa.ErrClusterDoesNotOwnShard: - http.Error(w, err.Error(), http.StatusPreconditionFailed) - default: - http.Error(w, err.Error(), http.StatusInternalServerError) + // Unmarshal request based on field type. + if field.Type() == pilosa.FieldTypeInt { + // Field type: Int + // Marshal into request object. + var req internal.ImportValueRequest + if err := proto.Unmarshal(body, &req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := h.API.ImportValue(r.Context(), req); err != nil { + switch errors.Cause(err) { + case pilosa.ErrClusterDoesNotOwnShard: + http.Error(w, err.Error(), http.StatusPreconditionFailed) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + } else { + // Field type: Set, Time + // Marshal into request object. + var req internal.ImportRequest + if err := proto.Unmarshal(body, &req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := h.API.Import(r.Context(), req); err != nil { + switch errors.Cause(err) { + case pilosa.ErrClusterDoesNotOwnShard: + http.Error(w, err.Error(), http.StatusPreconditionFailed) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return } - return } // Marshal response object. - buf, e := proto.Marshal(&internal.ImportResponse{Err: errorString(err)}) + buf, e := proto.Marshal(&internal.ImportResponse{Err: ""}) if e != nil { - http.Error(w, fmt.Sprintf("marshal import response: %s", err), http.StatusInternalServerError) + http.Error(w, fmt.Sprintf("marshal import response"), http.StatusInternalServerError) return } // Write response. - if err != nil { - w.WriteHeader(http.StatusInternalServerError) - } w.Write(buf) }