mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 17:15:56 +00:00
Merge branch 'develop' into clearbit-notime
This commit is contained in:
Todd Gruben
commit
ddf1ffc797
11 changed files with 362 additions and 324 deletions
50
api.go
50
api.go
|
|
@ -35,18 +35,11 @@ import (
|
|||
// API provides the top level programmatic interface to Pilosa. It is usually
|
||||
// wrapped by a handler which provides an external interface (e.g. HTTP).
|
||||
type API struct {
|
||||
Holder *Holder
|
||||
// The execution engine for running queries.
|
||||
Executor interface {
|
||||
Execute(context context.Context, index string, query *pql.Query, slices []uint64, opt *ExecOptions) ([]interface{}, error)
|
||||
}
|
||||
Broadcaster Broadcaster
|
||||
BroadcastHandler BroadcastHandler
|
||||
StatusHandler StatusHandler
|
||||
Cluster *Cluster
|
||||
TranslateStore TranslateStore
|
||||
Logger Logger
|
||||
server *Server
|
||||
Holder *Holder
|
||||
Broadcaster Broadcaster
|
||||
Cluster *Cluster
|
||||
TranslateStore TranslateStore
|
||||
server *Server
|
||||
}
|
||||
|
||||
// APIOption is a functional option type for pilosa.API
|
||||
|
|
@ -55,14 +48,10 @@ type APIOption func(*API) error
|
|||
func OptAPIServer(s *Server) APIOption {
|
||||
return func(a *API) error {
|
||||
a.server = s
|
||||
a.Executor = s.executor
|
||||
a.TranslateStore = s.translateFile
|
||||
a.Holder = s.holder
|
||||
a.Broadcaster = s
|
||||
a.BroadcastHandler = s
|
||||
a.StatusHandler = s
|
||||
a.Cluster = s.Cluster
|
||||
a.Logger = s.logger
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
|
@ -73,7 +62,6 @@ func NewAPI(opts ...APIOption) (*API, error) {
|
|||
Broadcaster: NopBroadcaster,
|
||||
//BroadcastHandler: NopBroadcastHandler, // TODO: implement the nop
|
||||
//StatusHandler: NopStatusHandler, // TODO: implement the nop
|
||||
Logger: NopLogger,
|
||||
}
|
||||
|
||||
for _, opt := range opts {
|
||||
|
|
@ -129,7 +117,7 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
|
|||
ExcludeRowAttrs: req.ExcludeRowAttrs,
|
||||
ExcludeColumns: req.ExcludeColumns,
|
||||
}
|
||||
results, err := api.Executor.Execute(ctx, req.Index, q, req.Slices, execOpts)
|
||||
results, err := api.server.executor.Execute(ctx, req.Index, q, req.Slices, execOpts)
|
||||
if err != nil {
|
||||
return resp, errors.Wrap(err, "executing")
|
||||
}
|
||||
|
|
@ -210,7 +198,7 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
|
|||
Meta: options.Encode(),
|
||||
})
|
||||
if err != nil {
|
||||
api.Logger.Printf("problem sending CreateIndex message: %s", err)
|
||||
api.server.logger.Printf("problem sending CreateIndex message: %s", err)
|
||||
return nil, errors.Wrap(err, "sending CreateIndex message")
|
||||
}
|
||||
api.Holder.Stats.Count("createIndex", 1, 1.0)
|
||||
|
|
@ -248,7 +236,7 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
|
|||
Index: indexName,
|
||||
})
|
||||
if err != nil {
|
||||
api.Logger.Printf("problem sending DeleteIndex message: %s", err)
|
||||
api.server.logger.Printf("problem sending DeleteIndex message: %s", err)
|
||||
return errors.Wrap(err, "sending DeleteIndex message")
|
||||
}
|
||||
api.Holder.Stats.Count("deleteIndex", 1, 1.0)
|
||||
|
|
@ -281,7 +269,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
|
|||
Meta: options.Encode(),
|
||||
})
|
||||
if err != nil {
|
||||
api.Logger.Printf("problem sending CreateField message: %s", err)
|
||||
api.server.logger.Printf("problem sending CreateField message: %s", err)
|
||||
return nil, errors.Wrap(err, "sending CreateField message")
|
||||
}
|
||||
api.Holder.Stats.CountWithCustomTags("createField", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)})
|
||||
|
|
@ -314,7 +302,7 @@ func (api *API) DeleteField(ctx context.Context, indexName string, fieldName str
|
|||
Field: fieldName,
|
||||
})
|
||||
if err != nil {
|
||||
api.Logger.Printf("problem sending DeleteField message: %s", err)
|
||||
api.server.logger.Printf("problem sending DeleteField message: %s", err)
|
||||
return errors.Wrap(err, "sending DeleteField message")
|
||||
}
|
||||
api.Holder.Stats.CountWithCustomTags("deleteField", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)})
|
||||
|
|
@ -330,7 +318,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
|
|||
|
||||
// Validate that this handler owns the slice.
|
||||
if !api.Cluster.ownsSlice(api.LocalID(), indexName, slice) {
|
||||
api.Logger.Printf("node %s does not own slice %d of index %s", api.LocalID(), slice, indexName)
|
||||
api.server.logger.Printf("node %s does not own slice %d of index %s", api.LocalID(), slice, indexName)
|
||||
return ErrClusterDoesNotOwnSlice
|
||||
}
|
||||
|
||||
|
|
@ -509,7 +497,7 @@ func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error {
|
|||
}
|
||||
|
||||
// Forward the error message.
|
||||
if err := api.BroadcastHandler.ReceiveMessage(pb); err != nil {
|
||||
if err := api.server.ReceiveMessage(pb); err != nil {
|
||||
return errors.Wrap(err, "receiving message")
|
||||
}
|
||||
return nil
|
||||
|
|
@ -571,7 +559,7 @@ func (api *API) DeleteView(ctx context.Context, indexName string, fieldName stri
|
|||
View: viewName,
|
||||
})
|
||||
if err != nil {
|
||||
api.Logger.Printf("problem sending DeleteView message: %s", err)
|
||||
api.server.logger.Printf("problem sending DeleteView message: %s", err)
|
||||
}
|
||||
|
||||
return errors.Wrap(err, "sending DeleteView message")
|
||||
|
|
@ -670,7 +658,7 @@ func (api *API) Import(ctx context.Context, req internal.ImportRequest) error {
|
|||
// Import into fragment.
|
||||
err = field.Import(req.RowIDs, req.ColumnIDs, timestamps)
|
||||
if err != nil {
|
||||
api.Logger.Printf("import error: index=%s, field=%s, slice=%d, columns=%d, err=%s", req.Index, req.Field, req.Slice, len(req.ColumnIDs), err)
|
||||
api.server.logger.Printf("import error: index=%s, field=%s, slice=%d, columns=%d, err=%s", req.Index, req.Field, req.Slice, len(req.ColumnIDs), err)
|
||||
}
|
||||
return errors.Wrap(err, "importing")
|
||||
}
|
||||
|
|
@ -688,7 +676,7 @@ func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest
|
|||
// Import into fragment.
|
||||
err = field.ImportValue(req.ColumnIDs, req.Values)
|
||||
if err != nil {
|
||||
api.Logger.Printf("import error: index=%s, field=%s, slice=%d, columns=%d, err=%s", req.Index, req.Field, req.Slice, len(req.ColumnIDs), err)
|
||||
api.server.logger.Printf("import error: index=%s, field=%s, slice=%d, columns=%d, err=%s", req.Index, req.Field, req.Slice, len(req.ColumnIDs), err)
|
||||
}
|
||||
return errors.Wrap(err, "importing")
|
||||
}
|
||||
|
|
@ -719,22 +707,22 @@ func (api *API) LongQueryTime() time.Duration {
|
|||
func (api *API) indexField(indexName string, fieldName string, slice uint64) (*Index, *Field, error) {
|
||||
// Validate that this handler owns the slice.
|
||||
if !api.Cluster.ownsSlice(api.LocalID(), indexName, slice) {
|
||||
api.Logger.Printf("node %s does not own slice %d of index %s", api.LocalID(), slice, indexName)
|
||||
api.server.logger.Printf("node %s does not own slice %d of index %s", api.LocalID(), slice, indexName)
|
||||
return nil, nil, ErrClusterDoesNotOwnSlice
|
||||
}
|
||||
|
||||
// Find the Index.
|
||||
api.Logger.Printf("importing: %v %v %v", indexName, fieldName, slice)
|
||||
api.server.logger.Printf("importing: %v %v %v", indexName, fieldName, slice)
|
||||
index := api.Holder.Index(indexName)
|
||||
if index == nil {
|
||||
api.Logger.Printf("fragment error: index=%s, field=%s, slice=%d, err=%s", indexName, fieldName, slice, ErrIndexNotFound.Error())
|
||||
api.server.logger.Printf("fragment error: index=%s, field=%s, slice=%d, err=%s", indexName, fieldName, slice, ErrIndexNotFound.Error())
|
||||
return nil, nil, ErrIndexNotFound
|
||||
}
|
||||
|
||||
// Retrieve field.
|
||||
field := index.Field(fieldName)
|
||||
if field == nil {
|
||||
api.Logger.Printf("field error: index=%s, field=%s, slice=%d, err=%s", indexName, fieldName, slice, ErrFieldNotFound.Error())
|
||||
api.server.logger.Printf("field error: index=%s, field=%s, slice=%d, err=%s", indexName, fieldName, slice, ErrFieldNotFound.Error())
|
||||
return nil, nil, ErrFieldNotFound
|
||||
}
|
||||
return index, field, nil
|
||||
|
|
|
|||
|
|
@ -283,24 +283,24 @@ func (h *Handler) handleHome(w http.ResponseWriter, r *http.Request) {
|
|||
http.Error(w, "Welcome. Pilosa is running. Visit https://www.pilosa.com/docs/ for more information.", http.StatusNotFound)
|
||||
}
|
||||
|
||||
func checkHeaderAcceptJSON(header http.Header) bool {
|
||||
v, found := header["Accept"]
|
||||
sendError := false
|
||||
if found {
|
||||
sendError = true
|
||||
// validHeaderAcceptJSON returns false if one or more Accept
|
||||
// headers are present, but none of them are "application/json"
|
||||
// (or any matching wildcard). Otherwise returns true.
|
||||
func validHeaderAcceptJSON(header http.Header) bool {
|
||||
if v, found := header["Accept"]; found {
|
||||
for _, v := range v {
|
||||
if v == "application/json" {
|
||||
sendError = false
|
||||
|
||||
if v == "application/json" || v == "*/*" || v == "*/json" || v == "application/*" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
return sendError
|
||||
return true
|
||||
}
|
||||
|
||||
// handleGetSchema handles GET /schema requests.
|
||||
func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -315,7 +315,7 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
// handleGetStatus handles GET /status requests.
|
||||
func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -330,7 +330,7 @@ func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
|
||||
func (h *Handler) handleGetInfo(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -387,7 +387,7 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
// handleGetSlicesMax handles GET /schema requests.
|
||||
func (h *Handler) handleGetSlicesMax(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -409,7 +409,7 @@ func (h *Handler) handleGetIndexes(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
// handleGetIndex handles GET /index/<indexname> requests.
|
||||
func (h *Handler) handleGetIndex(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -491,7 +491,7 @@ type postIndexResponse struct{}
|
|||
|
||||
// handleDeleteIndex handles DELETE /index request.
|
||||
func (h *Handler) handleDeleteIndex(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -513,7 +513,7 @@ type deleteIndexResponse struct{}
|
|||
|
||||
// handlePostIndex handles POST /index request.
|
||||
func (h *Handler) handlePostIndex(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -547,7 +547,7 @@ func (h *Handler) handlePostIndex(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
// handlePostIndexAttrDiff handles POST /index/attr/diff requests.
|
||||
func (h *Handler) handlePostIndexAttrDiff(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -588,7 +588,7 @@ type postIndexAttrDiffResponse struct {
|
|||
|
||||
// handlePostField handles POST /field request.
|
||||
func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -670,7 +670,7 @@ type postFieldResponse struct{}
|
|||
|
||||
// handleDeleteField handles DELETE /field request.
|
||||
func (h *Handler) handleDeleteField(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -700,7 +700,7 @@ type deleteFieldResponse struct{}
|
|||
|
||||
// handlePostFieldAttrDiff handles POST /field/attr/diff requests.
|
||||
func (h *Handler) handlePostFieldAttrDiff(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -796,7 +796,7 @@ func (h *Handler) readURLQueryRequest(r *http.Request) (*pilosa.QueryRequest, er
|
|||
|
||||
// writeQueryResponse writes the response from the executor to w.
|
||||
func (h *Handler) writeQueryResponse(w http.ResponseWriter, r *http.Request, resp *pilosa.QueryResponse) error {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
return h.writeProtobufQueryResponse(w, resp)
|
||||
}
|
||||
return h.writeJSONQueryResponse(w, resp)
|
||||
|
|
@ -959,7 +959,7 @@ func (h *Handler) handleGetExportCSV(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
// handleGetFragmentNodes handles /fragment/nodes requests.
|
||||
func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -1008,7 +1008,7 @@ func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Requ
|
|||
|
||||
// handleGetFragmentBlocks handles GET /fragment/blocks requests.
|
||||
func (h *Handler) handleGetFragmentBlocks(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -1044,7 +1044,7 @@ type getFragmentBlocksResponse struct {
|
|||
|
||||
// handleGetVersion handles /version requests.
|
||||
func (h *Handler) handleGetVersion(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -1146,7 +1146,7 @@ func errorString(err error) string {
|
|||
}
|
||||
|
||||
func (h *Handler) handlePostClusterResizeSetCoordinator(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -1187,7 +1187,7 @@ type setCoordinatorResponse struct {
|
|||
|
||||
// handlePostClusterResizeRemoveNode handles POST /cluster/resize/remove-node request.
|
||||
func (h *Handler) handlePostClusterResizeRemoveNode(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -1227,7 +1227,7 @@ type removeNodeResponse struct {
|
|||
|
||||
// handlePostClusterResizeAbort handles POST /cluster/resize/abort request.
|
||||
func (h *Handler) handlePostClusterResizeAbort(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
@ -1268,7 +1268,7 @@ func (h *Handler) handleRecalculateCaches(w http.ResponseWriter, r *http.Request
|
|||
}
|
||||
|
||||
func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Request) {
|
||||
if checkHeaderAcceptJSON(r.Header) {
|
||||
if !validHeaderAcceptJSON(r.Header) {
|
||||
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
|
||||
return
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,13 +4,13 @@ import (
|
|||
"context"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/pilosa/pilosa"
|
||||
"github.com/pilosa/pilosa/http"
|
||||
"github.com/pilosa/pilosa/mock"
|
||||
"github.com/pilosa/pilosa/server"
|
||||
"github.com/pilosa/pilosa/test"
|
||||
)
|
||||
|
||||
|
|
@ -52,13 +52,14 @@ func TestTranslateStore_Reader(t *testing.T) {
|
|||
}
|
||||
return &mrc, nil
|
||||
}
|
||||
h := test.MustNewHandler()
|
||||
h.API.TranslateStore = &translateStore
|
||||
s := httptest.NewServer(h)
|
||||
defer s.Close()
|
||||
|
||||
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
|
||||
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
|
||||
defer main.Close()
|
||||
|
||||
// Connect to server and stream all available data.
|
||||
store := http.NewTranslateStore(s.URL)
|
||||
store := http.NewTranslateStore(main.Server.URI.String())
|
||||
|
||||
rc, err := store.Reader(context.Background(), 100)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -95,15 +96,16 @@ func TestTranslateStore_Reader(t *testing.T) {
|
|||
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
|
||||
return &mrc, nil
|
||||
}
|
||||
h := test.MustNewHandler()
|
||||
h.API.TranslateStore = &translateStore
|
||||
s := httptest.NewServer(h)
|
||||
defer s.Close()
|
||||
|
||||
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
|
||||
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
|
||||
|
||||
defer main.Close()
|
||||
defer close(done)
|
||||
|
||||
// Connect to server and begin streaming.
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
store := http.NewTranslateStore(s.URL)
|
||||
store := http.NewTranslateStore(main.Server.URI.String())
|
||||
if _, err := store.Reader(ctx, 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -123,12 +125,11 @@ func TestTranslateStore_Reader(t *testing.T) {
|
|||
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
|
||||
return nil, pilosa.ErrNotImplemented
|
||||
}
|
||||
h := test.MustNewHandler()
|
||||
h.API.TranslateStore = &translateStore
|
||||
s := httptest.NewServer(h)
|
||||
defer s.Close()
|
||||
|
||||
_, err := http.NewTranslateStore(s.URL).Reader(context.Background(), 0)
|
||||
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
|
||||
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
|
||||
|
||||
_, err := http.NewTranslateStore(main.Server.URI.String()).Reader(context.Background(), 0)
|
||||
if err != pilosa.ErrNotImplemented {
|
||||
t.Fatalf("unexpected error: %s", err)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -17,22 +17,22 @@ type TranslateStore struct {
|
|||
ReaderFunc func(ctx context.Context, off int64) (io.ReadCloser, error)
|
||||
}
|
||||
|
||||
func (s *TranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
|
||||
func (s TranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
|
||||
return s.TranslateColumnsToUint64Func(index, values)
|
||||
}
|
||||
|
||||
func (s *TranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
|
||||
func (s TranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
|
||||
return s.TranslateColumnToStringFunc(index, values)
|
||||
}
|
||||
|
||||
func (s *TranslateStore) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
|
||||
func (s TranslateStore) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
|
||||
return s.TranslateRowsToUint64Func(index, frame, values)
|
||||
}
|
||||
|
||||
func (s *TranslateStore) TranslateRowToString(index, frame string, value uint64) (string, error) {
|
||||
func (s TranslateStore) TranslateRowToString(index, frame string, value uint64) (string, error) {
|
||||
return s.TranslateRowToStringFunc(index, frame, value)
|
||||
}
|
||||
|
||||
func (s *TranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
|
||||
func (s TranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
|
||||
return s.ReaderFunc(ctx, off)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ item <- ( 'null' &(comma / sp close) { p.addVal(nil) }
|
|||
doublequotedstring <- ( [^"\\\n] / '\\n' / '\\\"' / '\\\'' / '\\\\' )*
|
||||
singlequotedstring <- ( [^'\\\n] / '\\n' / '\\\"' / '\\\'' / '\\\\' )*
|
||||
|
||||
fieldExpr <- [[A-Z]] ( [[A-Z]] / [0-9] / '_' )*
|
||||
fieldExpr <- [[A-Z]] ( [[A-Z]] / [0-9] / '_' / '-' )*
|
||||
field <- <fieldExpr / reserved> { p.addField(buffer[begin:end]) }
|
||||
reserved <- ('_row' / '_col' / '_start' / '_end' / '_timestamp' / '_field')
|
||||
posfield <- <fieldExpr> { p.addPosStr("_field", buffer[begin:end]) }
|
||||
|
|
|
|||
439
pql/pql.peg.go
439
pql/pql.peg.go
|
|
@ -2118,7 +2118,7 @@ func (p *PQL) Init() {
|
|||
},
|
||||
/* 15 singlequotedstring <- <((!('\'' / '\\' / '\n') .) / ('\\' 'n') / ('\\' '"') / ('\\' '\'') / ('\\' '\\'))*> */
|
||||
nil,
|
||||
/* 16 fieldExpr <- <(([a-z] / [A-Z]) ([a-z] / [A-Z] / [0-9] / '_')*)> */
|
||||
/* 16 fieldExpr <- <(([a-z] / [A-Z]) ([a-z] / [A-Z] / [0-9] / '_' / '-')*)> */
|
||||
func() bool {
|
||||
position211, tokenIndex211 := position, tokenIndex
|
||||
{
|
||||
|
|
@ -2165,6 +2165,13 @@ func (p *PQL) Init() {
|
|||
l220:
|
||||
position, tokenIndex = position217, tokenIndex217
|
||||
if buffer[position] != rune('_') {
|
||||
goto l221
|
||||
}
|
||||
position++
|
||||
goto l217
|
||||
l221:
|
||||
position, tokenIndex = position217, tokenIndex217
|
||||
if buffer[position] != rune('-') {
|
||||
goto l216
|
||||
}
|
||||
position++
|
||||
|
|
@ -2183,391 +2190,391 @@ func (p *PQL) Init() {
|
|||
},
|
||||
/* 17 field <- <(<(fieldExpr / reserved)> Action38)> */
|
||||
func() bool {
|
||||
position221, tokenIndex221 := position, tokenIndex
|
||||
position222, tokenIndex222 := position, tokenIndex
|
||||
{
|
||||
position222 := position
|
||||
position223 := position
|
||||
{
|
||||
position223 := position
|
||||
position224 := position
|
||||
{
|
||||
position224, tokenIndex224 := position, tokenIndex
|
||||
position225, tokenIndex225 := position, tokenIndex
|
||||
if !_rules[rulefieldExpr]() {
|
||||
goto l225
|
||||
goto l226
|
||||
}
|
||||
goto l224
|
||||
l225:
|
||||
position, tokenIndex = position224, tokenIndex224
|
||||
goto l225
|
||||
l226:
|
||||
position, tokenIndex = position225, tokenIndex225
|
||||
{
|
||||
position226 := position
|
||||
position227 := position
|
||||
{
|
||||
position227, tokenIndex227 := position, tokenIndex
|
||||
position228, tokenIndex228 := position, tokenIndex
|
||||
if buffer[position] != rune('_') {
|
||||
goto l228
|
||||
goto l229
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('r') {
|
||||
goto l228
|
||||
goto l229
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('o') {
|
||||
goto l228
|
||||
goto l229
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('w') {
|
||||
goto l228
|
||||
goto l229
|
||||
}
|
||||
position++
|
||||
goto l227
|
||||
l228:
|
||||
position, tokenIndex = position227, tokenIndex227
|
||||
goto l228
|
||||
l229:
|
||||
position, tokenIndex = position228, tokenIndex228
|
||||
if buffer[position] != rune('_') {
|
||||
goto l229
|
||||
goto l230
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('c') {
|
||||
goto l229
|
||||
goto l230
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('o') {
|
||||
goto l229
|
||||
goto l230
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('l') {
|
||||
goto l229
|
||||
goto l230
|
||||
}
|
||||
position++
|
||||
goto l227
|
||||
l229:
|
||||
position, tokenIndex = position227, tokenIndex227
|
||||
goto l228
|
||||
l230:
|
||||
position, tokenIndex = position228, tokenIndex228
|
||||
if buffer[position] != rune('_') {
|
||||
goto l230
|
||||
goto l231
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('s') {
|
||||
goto l230
|
||||
goto l231
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('t') {
|
||||
goto l230
|
||||
goto l231
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('a') {
|
||||
goto l230
|
||||
goto l231
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('r') {
|
||||
goto l230
|
||||
goto l231
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('t') {
|
||||
goto l230
|
||||
goto l231
|
||||
}
|
||||
position++
|
||||
goto l227
|
||||
l230:
|
||||
position, tokenIndex = position227, tokenIndex227
|
||||
goto l228
|
||||
l231:
|
||||
position, tokenIndex = position228, tokenIndex228
|
||||
if buffer[position] != rune('_') {
|
||||
goto l231
|
||||
goto l232
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('e') {
|
||||
goto l231
|
||||
goto l232
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('n') {
|
||||
goto l231
|
||||
goto l232
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('d') {
|
||||
goto l231
|
||||
goto l232
|
||||
}
|
||||
position++
|
||||
goto l227
|
||||
l231:
|
||||
position, tokenIndex = position227, tokenIndex227
|
||||
goto l228
|
||||
l232:
|
||||
position, tokenIndex = position228, tokenIndex228
|
||||
if buffer[position] != rune('_') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('t') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('i') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('m') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('e') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('s') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('t') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('a') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('m') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('p') {
|
||||
goto l232
|
||||
goto l233
|
||||
}
|
||||
position++
|
||||
goto l227
|
||||
l232:
|
||||
position, tokenIndex = position227, tokenIndex227
|
||||
goto l228
|
||||
l233:
|
||||
position, tokenIndex = position228, tokenIndex228
|
||||
if buffer[position] != rune('_') {
|
||||
goto l221
|
||||
goto l222
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('f') {
|
||||
goto l221
|
||||
goto l222
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('i') {
|
||||
goto l221
|
||||
goto l222
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('e') {
|
||||
goto l221
|
||||
goto l222
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('l') {
|
||||
goto l221
|
||||
goto l222
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('d') {
|
||||
goto l221
|
||||
goto l222
|
||||
}
|
||||
position++
|
||||
}
|
||||
l227:
|
||||
add(rulereserved, position226)
|
||||
l228:
|
||||
add(rulereserved, position227)
|
||||
}
|
||||
}
|
||||
l224:
|
||||
add(rulePegText, position223)
|
||||
l225:
|
||||
add(rulePegText, position224)
|
||||
}
|
||||
{
|
||||
add(ruleAction38, position)
|
||||
}
|
||||
add(rulefield, position222)
|
||||
add(rulefield, position223)
|
||||
}
|
||||
return true
|
||||
l221:
|
||||
position, tokenIndex = position221, tokenIndex221
|
||||
l222:
|
||||
position, tokenIndex = position222, tokenIndex222
|
||||
return false
|
||||
},
|
||||
/* 18 reserved <- <(('_' 'r' 'o' 'w') / ('_' 'c' 'o' 'l') / ('_' 's' 't' 'a' 'r' 't') / ('_' 'e' 'n' 'd') / ('_' 't' 'i' 'm' 'e' 's' 't' 'a' 'm' 'p') / ('_' 'f' 'i' 'e' 'l' 'd'))> */
|
||||
nil,
|
||||
/* 19 posfield <- <(<fieldExpr> Action39)> */
|
||||
func() bool {
|
||||
position235, tokenIndex235 := position, tokenIndex
|
||||
position236, tokenIndex236 := position, tokenIndex
|
||||
{
|
||||
position236 := position
|
||||
position237 := position
|
||||
{
|
||||
position237 := position
|
||||
position238 := position
|
||||
if !_rules[rulefieldExpr]() {
|
||||
goto l235
|
||||
goto l236
|
||||
}
|
||||
add(rulePegText, position237)
|
||||
add(rulePegText, position238)
|
||||
}
|
||||
{
|
||||
add(ruleAction39, position)
|
||||
}
|
||||
add(ruleposfield, position236)
|
||||
add(ruleposfield, position237)
|
||||
}
|
||||
return true
|
||||
l235:
|
||||
position, tokenIndex = position235, tokenIndex235
|
||||
l236:
|
||||
position, tokenIndex = position236, tokenIndex236
|
||||
return false
|
||||
},
|
||||
/* 20 uint <- <(([1-9] [0-9]*) / '0')> */
|
||||
func() bool {
|
||||
position239, tokenIndex239 := position, tokenIndex
|
||||
position240, tokenIndex240 := position, tokenIndex
|
||||
{
|
||||
position240 := position
|
||||
position241 := position
|
||||
{
|
||||
position241, tokenIndex241 := position, tokenIndex
|
||||
position242, tokenIndex242 := position, tokenIndex
|
||||
if c := buffer[position]; c < rune('1') || c > rune('9') {
|
||||
goto l242
|
||||
goto l243
|
||||
}
|
||||
position++
|
||||
l243:
|
||||
l244:
|
||||
{
|
||||
position244, tokenIndex244 := position, tokenIndex
|
||||
position245, tokenIndex245 := position, tokenIndex
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l244
|
||||
goto l245
|
||||
}
|
||||
position++
|
||||
goto l243
|
||||
l244:
|
||||
position, tokenIndex = position244, tokenIndex244
|
||||
goto l244
|
||||
l245:
|
||||
position, tokenIndex = position245, tokenIndex245
|
||||
}
|
||||
goto l241
|
||||
l242:
|
||||
position, tokenIndex = position241, tokenIndex241
|
||||
goto l242
|
||||
l243:
|
||||
position, tokenIndex = position242, tokenIndex242
|
||||
if buffer[position] != rune('0') {
|
||||
goto l239
|
||||
goto l240
|
||||
}
|
||||
position++
|
||||
}
|
||||
l241:
|
||||
add(ruleuint, position240)
|
||||
l242:
|
||||
add(ruleuint, position241)
|
||||
}
|
||||
return true
|
||||
l239:
|
||||
position, tokenIndex = position239, tokenIndex239
|
||||
l240:
|
||||
position, tokenIndex = position240, tokenIndex240
|
||||
return false
|
||||
},
|
||||
/* 21 uintrow <- <(<uint> Action40)> */
|
||||
nil,
|
||||
/* 22 col <- <((<uint> Action41) / ('"' <doublequotedstring> '"' Action42))> */
|
||||
func() bool {
|
||||
position246, tokenIndex246 := position, tokenIndex
|
||||
position247, tokenIndex247 := position, tokenIndex
|
||||
{
|
||||
position247 := position
|
||||
position248 := position
|
||||
{
|
||||
position248, tokenIndex248 := position, tokenIndex
|
||||
position249, tokenIndex249 := position, tokenIndex
|
||||
{
|
||||
position250 := position
|
||||
position251 := position
|
||||
if !_rules[ruleuint]() {
|
||||
goto l249
|
||||
goto l250
|
||||
}
|
||||
add(rulePegText, position250)
|
||||
add(rulePegText, position251)
|
||||
}
|
||||
{
|
||||
add(ruleAction41, position)
|
||||
}
|
||||
goto l248
|
||||
l249:
|
||||
position, tokenIndex = position248, tokenIndex248
|
||||
goto l249
|
||||
l250:
|
||||
position, tokenIndex = position249, tokenIndex249
|
||||
if buffer[position] != rune('"') {
|
||||
goto l246
|
||||
goto l247
|
||||
}
|
||||
position++
|
||||
{
|
||||
position252 := position
|
||||
position253 := position
|
||||
if !_rules[ruledoublequotedstring]() {
|
||||
goto l246
|
||||
goto l247
|
||||
}
|
||||
add(rulePegText, position252)
|
||||
add(rulePegText, position253)
|
||||
}
|
||||
if buffer[position] != rune('"') {
|
||||
goto l246
|
||||
goto l247
|
||||
}
|
||||
position++
|
||||
{
|
||||
add(ruleAction42, position)
|
||||
}
|
||||
}
|
||||
l248:
|
||||
add(rulecol, position247)
|
||||
l249:
|
||||
add(rulecol, position248)
|
||||
}
|
||||
return true
|
||||
l246:
|
||||
position, tokenIndex = position246, tokenIndex246
|
||||
l247:
|
||||
position, tokenIndex = position247, tokenIndex247
|
||||
return false
|
||||
},
|
||||
/* 23 open <- <('(' sp)> */
|
||||
func() bool {
|
||||
position254, tokenIndex254 := position, tokenIndex
|
||||
position255, tokenIndex255 := position, tokenIndex
|
||||
{
|
||||
position255 := position
|
||||
position256 := position
|
||||
if buffer[position] != rune('(') {
|
||||
goto l254
|
||||
goto l255
|
||||
}
|
||||
position++
|
||||
if !_rules[rulesp]() {
|
||||
goto l254
|
||||
goto l255
|
||||
}
|
||||
add(ruleopen, position255)
|
||||
add(ruleopen, position256)
|
||||
}
|
||||
return true
|
||||
l254:
|
||||
position, tokenIndex = position254, tokenIndex254
|
||||
l255:
|
||||
position, tokenIndex = position255, tokenIndex255
|
||||
return false
|
||||
},
|
||||
/* 24 close <- <(')' sp)> */
|
||||
func() bool {
|
||||
position256, tokenIndex256 := position, tokenIndex
|
||||
position257, tokenIndex257 := position, tokenIndex
|
||||
{
|
||||
position257 := position
|
||||
position258 := position
|
||||
if buffer[position] != rune(')') {
|
||||
goto l256
|
||||
goto l257
|
||||
}
|
||||
position++
|
||||
if !_rules[rulesp]() {
|
||||
goto l256
|
||||
goto l257
|
||||
}
|
||||
add(ruleclose, position257)
|
||||
add(ruleclose, position258)
|
||||
}
|
||||
return true
|
||||
l256:
|
||||
position, tokenIndex = position256, tokenIndex256
|
||||
l257:
|
||||
position, tokenIndex = position257, tokenIndex257
|
||||
return false
|
||||
},
|
||||
/* 25 sp <- <(' ' / '\t')*> */
|
||||
func() bool {
|
||||
{
|
||||
position259 := position
|
||||
l260:
|
||||
position260 := position
|
||||
l261:
|
||||
{
|
||||
position261, tokenIndex261 := position, tokenIndex
|
||||
position262, tokenIndex262 := position, tokenIndex
|
||||
{
|
||||
position262, tokenIndex262 := position, tokenIndex
|
||||
position263, tokenIndex263 := position, tokenIndex
|
||||
if buffer[position] != rune(' ') {
|
||||
goto l263
|
||||
goto l264
|
||||
}
|
||||
position++
|
||||
goto l262
|
||||
l263:
|
||||
position, tokenIndex = position262, tokenIndex262
|
||||
goto l263
|
||||
l264:
|
||||
position, tokenIndex = position263, tokenIndex263
|
||||
if buffer[position] != rune('\t') {
|
||||
goto l261
|
||||
goto l262
|
||||
}
|
||||
position++
|
||||
}
|
||||
l263:
|
||||
goto l261
|
||||
l262:
|
||||
goto l260
|
||||
l261:
|
||||
position, tokenIndex = position261, tokenIndex261
|
||||
position, tokenIndex = position262, tokenIndex262
|
||||
}
|
||||
add(rulesp, position259)
|
||||
add(rulesp, position260)
|
||||
}
|
||||
return true
|
||||
},
|
||||
/* 26 comma <- <(sp ',' whitesp)> */
|
||||
func() bool {
|
||||
position264, tokenIndex264 := position, tokenIndex
|
||||
position265, tokenIndex265 := position, tokenIndex
|
||||
{
|
||||
position265 := position
|
||||
position266 := position
|
||||
if !_rules[rulesp]() {
|
||||
goto l264
|
||||
goto l265
|
||||
}
|
||||
if buffer[position] != rune(',') {
|
||||
goto l264
|
||||
goto l265
|
||||
}
|
||||
position++
|
||||
if !_rules[rulewhitesp]() {
|
||||
goto l264
|
||||
goto l265
|
||||
}
|
||||
add(rulecomma, position265)
|
||||
add(rulecomma, position266)
|
||||
}
|
||||
return true
|
||||
l264:
|
||||
position, tokenIndex = position264, tokenIndex264
|
||||
l265:
|
||||
position, tokenIndex = position265, tokenIndex265
|
||||
return false
|
||||
},
|
||||
/* 27 lbrack <- <('[' sp)> */
|
||||
|
|
@ -2577,37 +2584,37 @@ func (p *PQL) Init() {
|
|||
/* 29 whitesp <- <(' ' / '\t' / '\n')*> */
|
||||
func() bool {
|
||||
{
|
||||
position269 := position
|
||||
l270:
|
||||
position270 := position
|
||||
l271:
|
||||
{
|
||||
position271, tokenIndex271 := position, tokenIndex
|
||||
position272, tokenIndex272 := position, tokenIndex
|
||||
{
|
||||
position272, tokenIndex272 := position, tokenIndex
|
||||
position273, tokenIndex273 := position, tokenIndex
|
||||
if buffer[position] != rune(' ') {
|
||||
goto l273
|
||||
}
|
||||
position++
|
||||
goto l272
|
||||
l273:
|
||||
position, tokenIndex = position272, tokenIndex272
|
||||
if buffer[position] != rune('\t') {
|
||||
goto l274
|
||||
}
|
||||
position++
|
||||
goto l272
|
||||
goto l273
|
||||
l274:
|
||||
position, tokenIndex = position272, tokenIndex272
|
||||
position, tokenIndex = position273, tokenIndex273
|
||||
if buffer[position] != rune('\t') {
|
||||
goto l275
|
||||
}
|
||||
position++
|
||||
goto l273
|
||||
l275:
|
||||
position, tokenIndex = position273, tokenIndex273
|
||||
if buffer[position] != rune('\n') {
|
||||
goto l271
|
||||
goto l272
|
||||
}
|
||||
position++
|
||||
}
|
||||
l273:
|
||||
goto l271
|
||||
l272:
|
||||
goto l270
|
||||
l271:
|
||||
position, tokenIndex = position271, tokenIndex271
|
||||
position, tokenIndex = position272, tokenIndex272
|
||||
}
|
||||
add(rulewhitesp, position269)
|
||||
add(rulewhitesp, position270)
|
||||
}
|
||||
return true
|
||||
},
|
||||
|
|
@ -2615,136 +2622,136 @@ func (p *PQL) Init() {
|
|||
nil,
|
||||
/* 31 timestampbasicfmt <- <([0-9] [0-9] [0-9] [0-9] '-' ('0' / '1') [0-9] '-' [0-3] [0-9] 'T' [0-9] [0-9] ':' [0-9] [0-9])> */
|
||||
func() bool {
|
||||
position276, tokenIndex276 := position, tokenIndex
|
||||
position277, tokenIndex277 := position, tokenIndex
|
||||
{
|
||||
position277 := position
|
||||
position278 := position
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('-') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
{
|
||||
position278, tokenIndex278 := position, tokenIndex
|
||||
position279, tokenIndex279 := position, tokenIndex
|
||||
if buffer[position] != rune('0') {
|
||||
goto l279
|
||||
goto l280
|
||||
}
|
||||
position++
|
||||
goto l278
|
||||
l279:
|
||||
position, tokenIndex = position278, tokenIndex278
|
||||
goto l279
|
||||
l280:
|
||||
position, tokenIndex = position279, tokenIndex279
|
||||
if buffer[position] != rune('1') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
}
|
||||
l278:
|
||||
l279:
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('-') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if c := buffer[position]; c < rune('0') || c > rune('3') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune('T') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if buffer[position] != rune(':') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
if c := buffer[position]; c < rune('0') || c > rune('9') {
|
||||
goto l276
|
||||
goto l277
|
||||
}
|
||||
position++
|
||||
add(ruletimestampbasicfmt, position277)
|
||||
add(ruletimestampbasicfmt, position278)
|
||||
}
|
||||
return true
|
||||
l276:
|
||||
position, tokenIndex = position276, tokenIndex276
|
||||
l277:
|
||||
position, tokenIndex = position277, tokenIndex277
|
||||
return false
|
||||
},
|
||||
/* 32 timestampfmt <- <(('"' timestampbasicfmt '"') / ('\'' timestampbasicfmt '\'') / timestampbasicfmt)> */
|
||||
func() bool {
|
||||
position280, tokenIndex280 := position, tokenIndex
|
||||
position281, tokenIndex281 := position, tokenIndex
|
||||
{
|
||||
position281 := position
|
||||
position282 := position
|
||||
{
|
||||
position282, tokenIndex282 := position, tokenIndex
|
||||
position283, tokenIndex283 := position, tokenIndex
|
||||
if buffer[position] != rune('"') {
|
||||
goto l283
|
||||
}
|
||||
position++
|
||||
if !_rules[ruletimestampbasicfmt]() {
|
||||
goto l283
|
||||
}
|
||||
if buffer[position] != rune('"') {
|
||||
goto l283
|
||||
}
|
||||
position++
|
||||
goto l282
|
||||
l283:
|
||||
position, tokenIndex = position282, tokenIndex282
|
||||
if buffer[position] != rune('\'') {
|
||||
goto l284
|
||||
}
|
||||
position++
|
||||
if !_rules[ruletimestampbasicfmt]() {
|
||||
goto l284
|
||||
}
|
||||
if buffer[position] != rune('\'') {
|
||||
if buffer[position] != rune('"') {
|
||||
goto l284
|
||||
}
|
||||
position++
|
||||
goto l282
|
||||
goto l283
|
||||
l284:
|
||||
position, tokenIndex = position282, tokenIndex282
|
||||
position, tokenIndex = position283, tokenIndex283
|
||||
if buffer[position] != rune('\'') {
|
||||
goto l285
|
||||
}
|
||||
position++
|
||||
if !_rules[ruletimestampbasicfmt]() {
|
||||
goto l280
|
||||
goto l285
|
||||
}
|
||||
if buffer[position] != rune('\'') {
|
||||
goto l285
|
||||
}
|
||||
position++
|
||||
goto l283
|
||||
l285:
|
||||
position, tokenIndex = position283, tokenIndex283
|
||||
if !_rules[ruletimestampbasicfmt]() {
|
||||
goto l281
|
||||
}
|
||||
}
|
||||
l282:
|
||||
add(ruletimestampfmt, position281)
|
||||
l283:
|
||||
add(ruletimestampfmt, position282)
|
||||
}
|
||||
return true
|
||||
l280:
|
||||
position, tokenIndex = position280, tokenIndex280
|
||||
l281:
|
||||
position, tokenIndex = position281, tokenIndex281
|
||||
return false
|
||||
},
|
||||
/* 33 timestamp <- <(<timestampfmt> Action43)> */
|
||||
|
|
|
|||
|
|
@ -224,6 +224,10 @@ func TestPEGWorking(t *testing.T) {
|
|||
name: "RangeTimeQuotes",
|
||||
input: `Range(a=4, '2010-07-04T00:00', "2010-08-04T00:00")`,
|
||||
ncalls: 1},
|
||||
{
|
||||
name: "Dashed Frame",
|
||||
input: "Set(1, my-frame=9)",
|
||||
ncalls: 1},
|
||||
}
|
||||
|
||||
for i, test := range tests {
|
||||
|
|
@ -473,6 +477,15 @@ func TestPQLDeepEquality(t *testing.T) {
|
|||
"field": "f",
|
||||
},
|
||||
}},
|
||||
{
|
||||
name: "Weird dash",
|
||||
call: "Sum(field-=f)",
|
||||
exp: &Call{
|
||||
Name: "Sum",
|
||||
Args: map[string]interface{}{
|
||||
"field-": "f",
|
||||
},
|
||||
}},
|
||||
{
|
||||
name: "SumChild",
|
||||
call: "Sum(Row(), field=f)",
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ import (
|
|||
"github.com/pilosa/pilosa"
|
||||
"github.com/pilosa/pilosa/http"
|
||||
"github.com/pilosa/pilosa/internal"
|
||||
"github.com/pilosa/pilosa/server"
|
||||
"github.com/pilosa/pilosa/test"
|
||||
)
|
||||
|
||||
|
|
@ -565,7 +566,7 @@ func TestHandler_Endpoints(t *testing.T) {
|
|||
t.Fatalf("CORS preflight status should be 405, but is %v", result.StatusCode)
|
||||
}
|
||||
|
||||
clus := test.MustRunMainWithCluster(t, 1, test.OptAllowedOrigins([]string{"http://test/"}))
|
||||
clus := test.MustRunMainWithCluster(t, 1, []server.CommandOption{test.OptAllowedOrigins([]string{"http://test/"})})
|
||||
w = httptest.NewRecorder()
|
||||
h := clus[0].Handler.(*http.Handler).Handler
|
||||
h.ServeHTTP(w, req)
|
||||
|
|
|
|||
|
|
@ -76,11 +76,22 @@ type Command struct {
|
|||
|
||||
Handler pilosa.Handler
|
||||
ln net.Listener
|
||||
|
||||
serverOptions []pilosa.ServerOption
|
||||
}
|
||||
|
||||
type CommandOption func(c *Command) error
|
||||
|
||||
func OptCommandServerOptions(opts ...pilosa.ServerOption) CommandOption {
|
||||
return func(c *Command) error {
|
||||
c.serverOptions = append(c.serverOptions, opts...)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// NewCommand returns a new instance of Main.
|
||||
func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
|
||||
return &Command{
|
||||
func NewCommand(stdin io.Reader, stdout, stderr io.Writer, opts ...CommandOption) *Command {
|
||||
c := &Command{
|
||||
Config: NewConfig(),
|
||||
|
||||
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
|
||||
|
|
@ -88,6 +99,16 @@ func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
|
|||
Started: make(chan struct{}),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
|
||||
for _, opt := range opts {
|
||||
err := opt(c)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
// TODO: Return error instead of panic?
|
||||
}
|
||||
}
|
||||
|
||||
return c
|
||||
}
|
||||
|
||||
// Start starts the pilosa server - it returns once the server is running.
|
||||
|
|
@ -225,7 +246,7 @@ func (m *Command) SetupServer() error {
|
|||
primaryTranslateStore = http.NewTranslateStore(m.Config.Translation.PrimaryURL)
|
||||
}
|
||||
|
||||
m.Server, err = pilosa.NewServer(
|
||||
serverOptions := []pilosa.ServerOption{
|
||||
pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)),
|
||||
pilosa.OptServerLongQueryTime(time.Duration(m.Config.Cluster.LongQueryTime)),
|
||||
pilosa.OptServerDataDir(m.Config.DataDir),
|
||||
|
|
@ -243,7 +264,12 @@ func (m *Command) SetupServer() error {
|
|||
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
|
||||
pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore),
|
||||
pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts),
|
||||
)
|
||||
}
|
||||
|
||||
serverOptions = append(serverOptions, m.serverOptions...)
|
||||
|
||||
m.Server, err = pilosa.NewServer(serverOptions...)
|
||||
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "new server")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/pilosa/pilosa"
|
||||
"github.com/pilosa/pilosa/server"
|
||||
"github.com/pilosa/pilosa/test"
|
||||
)
|
||||
|
||||
|
|
@ -27,7 +28,7 @@ import (
|
|||
// pilosa.Server was not having its remoteClient field set by an option and so
|
||||
// it was using a nil client in monitorAntiEntropy.
|
||||
func TestMonitorAntiEntropy(t *testing.T) {
|
||||
cluster := test.MustRunMainWithCluster(t, 3, test.OptAntiEntropyInterval(time.Millisecond*20))
|
||||
cluster := test.MustRunMainWithCluster(t, 3, []server.CommandOption{test.OptAntiEntropyInterval(time.Millisecond * 20)})
|
||||
client := cluster[1].Client()
|
||||
err := client.CreateIndex(context.Background(), "balh", pilosa.IndexOptions{})
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -34,52 +34,46 @@ import (
|
|||
)
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////////////
|
||||
// Main represents a test wrapper for main.Main.
|
||||
// Main represents a test wrapper for server.Command.
|
||||
type Main struct {
|
||||
*server.Command
|
||||
|
||||
commandOptions []server.CommandOption
|
||||
|
||||
Stdin bytes.Buffer
|
||||
Stdout bytes.Buffer
|
||||
Stderr bytes.Buffer
|
||||
}
|
||||
|
||||
type MainOpt func(m *Main) error
|
||||
|
||||
func OptAntiEntropyInterval(dur time.Duration) MainOpt {
|
||||
return func(m *Main) error {
|
||||
m.Command.Config.AntiEntropy.Interval = toml.Duration(dur)
|
||||
func OptAntiEntropyInterval(dur time.Duration) server.CommandOption {
|
||||
return func(m *server.Command) error {
|
||||
m.Config.AntiEntropy.Interval = toml.Duration(dur)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func OptAllowedOrigins(origins []string) MainOpt {
|
||||
return func(m *Main) error {
|
||||
func OptAllowedOrigins(origins []string) server.CommandOption {
|
||||
return func(m *server.Command) error {
|
||||
m.Config.Handler.AllowedOrigins = origins
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// NewMain returns a new instance of Main with a temporary data directory and random port.
|
||||
func NewMain(opts ...MainOpt) *Main {
|
||||
func NewMain(opts ...server.CommandOption) *Main {
|
||||
path, err := ioutil.TempDir("", "pilosa-")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)}
|
||||
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr, opts...), commandOptions: opts}
|
||||
m.Config.DataDir = path
|
||||
m.Config.Bind = "http://localhost:0"
|
||||
m.Config.Cluster.Disabled = true
|
||||
m.Command.Stdin = &m.Stdin
|
||||
m.Command.Stdout = &m.Stdout
|
||||
m.Command.Stderr = &m.Stderr
|
||||
for _, opt := range opts {
|
||||
err := opt(m)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
}
|
||||
err = m.SetupServer()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
|
|
@ -94,7 +88,7 @@ func NewMain(opts ...MainOpt) *Main {
|
|||
}
|
||||
|
||||
// NewMainWithCluster returns a new instance of Main with clustering enabled.
|
||||
func NewMainWithCluster(isCoordinator bool, opts ...MainOpt) *Main {
|
||||
func NewMainWithCluster(isCoordinator bool, opts ...server.CommandOption) *Main {
|
||||
m := NewMain(opts...)
|
||||
m.Config.Cluster.Disabled = false
|
||||
m.Config.Cluster.Coordinator = isCoordinator
|
||||
|
|
@ -103,7 +97,7 @@ func NewMainWithCluster(isCoordinator bool, opts ...MainOpt) *Main {
|
|||
|
||||
// MustRunMainWithCluster ruturns a running array of *Main where
|
||||
// all nodes are joined via memberlist (i.e. clustering enabled).
|
||||
func MustRunMainWithCluster(t *testing.T, size int, opts ...MainOpt) []*Main {
|
||||
func MustRunMainWithCluster(t *testing.T, size int, opts ...[]server.CommandOption) []*Main {
|
||||
ma, err := runMainWithCluster(size, opts...)
|
||||
if err != nil {
|
||||
t.Fatalf("new main array with cluster: %v", err)
|
||||
|
|
@ -113,10 +107,13 @@ func MustRunMainWithCluster(t *testing.T, size int, opts ...MainOpt) []*Main {
|
|||
|
||||
// runMainWithCluster runs an array of *Main where all nodes are
|
||||
// joined via memberlist (i.e. clustering enabled).
|
||||
func runMainWithCluster(size int, opts ...MainOpt) ([]*Main, error) {
|
||||
func runMainWithCluster(size int, opts ...[]server.CommandOption) ([]*Main, error) {
|
||||
if size == 0 {
|
||||
return nil, errors.New("cluster must contain at least one node")
|
||||
}
|
||||
if len(opts) != size && len(opts) != 0 && len(opts) != 1 {
|
||||
return nil, errors.New("Slice of CommandOptions must be of length 0, 1, or equal to the number of cluster nodes")
|
||||
}
|
||||
|
||||
mains := make([]*Main, size)
|
||||
|
||||
|
|
@ -126,7 +123,11 @@ func runMainWithCluster(size int, opts ...MainOpt) ([]*Main, error) {
|
|||
var gossipSeeds = make([]string, size)
|
||||
|
||||
for i := 0; i < size; i++ {
|
||||
m := NewMainWithCluster(i == 0, opts...)
|
||||
var commandOpts []server.CommandOption
|
||||
if len(opts) > 0 {
|
||||
commandOpts = opts[i%len(opts)]
|
||||
}
|
||||
m := NewMainWithCluster(i == 0, commandOpts...)
|
||||
m.Config.Cluster.Disabled = false
|
||||
|
||||
gossipSeeds[i], err = m.RunWithTransport(gossipHost, gossipPort, gossipSeeds[:i])
|
||||
|
|
@ -164,7 +165,7 @@ func (m *Main) Reopen() error {
|
|||
|
||||
// Create new main with the same config.
|
||||
config := m.Command.Config
|
||||
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr)
|
||||
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr, m.commandOptions...)
|
||||
m.Command.Config = config
|
||||
err := m.SetupServer()
|
||||
if err != nil {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue